daniel/oshipka
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

acls

Browse Source
This commit is contained in:
Daniel Tsvetkov 2021-05-08 20:53:50 +02:00
parent e7a0d9f6bd
commit bad151eb43
8 changed files with 69 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
oshipka/persistance/__init__.py
View File

@ -34,6 +34,17 @@ db = SQLAlchemy()
migrate = Migrate()
SHARING_TYPE_TYPES_TYPE_PUBLIC = "PUBLIC"
SHARING_TYPE_TYPES_TYPE_AUTHZ = "AUTHZ"
SHARING_TYPE_TYPES_TYPE_AUTHN = "AUTHN"
SHARING_TYPE_TYPES = [
(SHARING_TYPE_TYPES_TYPE_PUBLIC, u'public'),
(SHARING_TYPE_TYPES_TYPE_AUTHZ, u'all logged in'),
(SHARING_TYPE_TYPES_TYPE_AUTHN, u'some authenticated users'),
]
class Ownable(object):
@declared_attr
def user_id(self):
@ -236,8 +247,28 @@ def register_filters(app):
def bool_filter(v):
return bool(v)
def check_instance_perm(model_view, verb, instance):
model_acl = model_view.model_acl
# Anonymous user -> check public ACL
if current_user.is_anonymous:
instance_acl = model_acl.query.filter_by(user=current_user, instance=instance,
acl_type=SHARING_TYPE_TYPES_TYPE_PUBLIC).first()
else:
# Logged in user -> find (user, instance) pair
instance_acl = model_acl.query.filter_by(user=current_user, instance=instance,
acl_type=SHARING_TYPE_TYPES_TYPE_AUTHZ).first()
if not instance_acl:
# If not (user, instance) pair -> check authN ACL
instance_acl = model_acl.query.filter_by(user=current_user, instance=instance,
acl_type=SHARING_TYPE_TYPES_TYPE_AUTHN).first()
column = verb.replace('.', '__')
return getattr(instance_acl, column)
def has_permission(model, verb, instance=None):
acl = MODEL_VIEWS.get(model, {}).model.acls.get(verb)
model_view = MODEL_VIEWS.get(model, {})
if '.' in verb:
return check_instance_perm(model_view, verb, instance)
acl = model_view.model.model_acls.get(verb)
# Anonymous user -> do we require AuthN?
if current_user.is_anonymous:
return not acl.get('authn')
@ -248,8 +279,6 @@ def register_filters(app):
return True
# One role is enough to grant permission
for role in roles:
if role in ['owner']:
return instance.user == current_user
if current_user.has_role(role):
return True
return False

14
oshipka/webapp/views.py
View File

@ -7,10 +7,12 @@ from functools import wraps
import inflect
from flask import flash, render_template, redirect, request, url_for, jsonify
from flask_login import current_user
from flask_security import login_required, roles_required
from sqlalchemy_filters import apply_filters
from oshipka.persistance import db, filter_m_n, update_m_ns
from oshipka.persistance import db, filter_m_n, update_m_ns, SHARING_TYPE_TYPES_TYPE_PUBLIC, \
SHARING_TYPE_TYPES_TYPE_AUTHZ, SHARING_TYPE_TYPES_TYPE_AUTHN
from oshipka.util.strings import camel_case_to_snake_case
from config import MEDIA_DIR
@ -128,6 +130,13 @@ def default_create_func(vc):
vc.instances = [instance]
default_update_func(vc)
instance_public_acl = vc.model_view.model_acl(user=current_user, instance=instance, acl_type=SHARING_TYPE_TYPES_TYPE_PUBLIC)
instance_authn_acl = vc.model_view.model_acl(user=current_user, instance=instance, acl_type=SHARING_TYPE_TYPES_TYPE_AUTHN)
instance_authz_acl = vc.model_view.model_acl(user=current_user, instance=instance, acl_type=SHARING_TYPE_TYPES_TYPE_AUTHZ)
db.session.add(instance_public_acl)
db.session.add(instance_authn_acl)
db.session.add(instance_authz_acl)
def default_delete_func(vc):
instance = vc.instances[0]
@ -248,9 +257,10 @@ def create_view(model_view, view_context_kwargs, is_login_required=False, the_ro
class ModelView(object):
def __init__(self, app, model):
def __init__(self, app, model, model_acl):
self.app = app
self.model = model
self.model_acl = model_acl
p = inflect.engine()
if hasattr(model, "__name__"):

5
vm_gen/templates/_model_py
View File

@ -7,9 +7,10 @@ class [[ name ]](db.Model, ModelController[% for inherit in interits %], [[ inhe
[%- include "_model_searchable_header_py" %]
_file_columns = [ [%- for column in columns %][%- if column.is_file %]"[[ column.name ]]", [%- endif %] [%- endfor %] ]
acls = [[ acls ]]
[%- for column in columns %]
model_acls = [[ acls ]]
[% for column in columns %]
[%- if column._type == 'relationship' %]
[%- include "_relationship_py" %]
[%- else %]

2
vm_gen/templates/html/_get.html
View File

@ -1,4 +1,5 @@
[%- for column in columns %]
{% if has_permission("[[ name|camel_to_snake ]]", "[[ column.name ]].read", instance) %}
{% if "[[ column.name ]]" not in skip_list %}
<li id="display-[[ name|camel_to_snake ]]-[[ column.name ]]"><strong>[%- if column.type in ['relationship'] and column.multiple %]{{ _("[[ column.name|pluralize ]]") }}[%- else %]{{ _("[[ column.name ]]") }}[%- endif %]</strong>:
[%- if not column.type in ['bool', 'boolean', ] %]
@ -37,6 +38,7 @@
[%- if not column.type in ['bool', 'boolean', ] %]
{% endif %}
[%- endif %]
{% endif %}
[%- endfor %]
[%- for backref in backrefs %]
<li id="display-[[ backref.name ]]"><strong>{{ _("[[ backref.name ]]") }}</strong>:

8
vm_gen/templates/html/_item.html Normal file
View File

@ -0,0 +1,8 @@
{% if has_permission('[[ name|camel_to_snake ]]', 'get') %}
<a href="{{ url_for('get_[[ name|camel_to_snake ]]', uuid=instance.id) }}">
{% endif %}
{% include "[[ name|camel_to_snake ]]/_title.html" %}
{% if has_permission('[[ name|camel_to_snake ]]', 'get') %}
</a>
{% endif %}
{% include "[[ name|camel_to_snake ]]/_actions.html" %}

9
vm_gen/templates/html/_list_item.html
View File

@ -1,10 +1,3 @@
<li>
{% if has_permission('[[ name|camel_to_snake ]]', 'get') %}
<a href="{{ url_for('get_[[ name|camel_to_snake ]]', uuid=instance.id) }}">
{% endif %}
{% include "[[ name|camel_to_snake ]]/_title.html" %}
{% if has_permission('[[ name|camel_to_snake ]]', 'get') %}
</a>
{% endif %}
{% include "[[ name|camel_to_snake ]]/_actions.html" %}
{% include "[[ name|camel_to_snake ]]/_item.html" %}
</li>

4
vm_gen/templates/routes_py
View File

@ -6,10 +6,10 @@ Edit the hooks in webapp/routes/[[ name|camel_to_snake ]]_hooks.py instead
from oshipka.webapp import app
from oshipka.webapp.views import ModelView
from webapp.models import [[ name ]]
from webapp.models import [[ name ]], [[ name ]]Acl
from webapp.routes.[[ name|camel_to_snake ]]_hooks import *
[[ name|camel_to_snake ]] = ModelView(app, [[name]])
[[ name|camel_to_snake ]] = ModelView(app, [[name]], [[ name ]]Acl)
[% for verb, verb_values in _verbs.items() %]
[[ name|camel_to_snake ]].register_verb(view_context=[[ verb ]]_view_context,
verb="[[ verb ]]",

12
vm_gen/vm_gen.py
View File

@ -165,14 +165,20 @@ def process_model(view_model):
model = autopep8.fix_code(template.render(**view_model), options=pep_options)
_model_name = view_model.get('name')
template_acl = env.get_template('model_acl_py')
model_acl = autopep8.fix_code(template_acl.render(**view_model), options=pep_options)
model_camel = _model_name.split('.yaml')[0]
model_snake = camel_case_to_snake_case(_model_name.split('.yaml')[0])
with open(os.path.join(MODELS_PATH, "_{}.py".format(model_snake)), 'w+') as f:
f.write(model)
with open(os.path.join(MODELS_PATH, "_{}_acl.py".format(model_snake)), 'w+') as f:
f.write(model_acl)
public_model = os.path.join(MODELS_PATH, "{}.py".format(model_snake))
if not os.path.exists(public_model):
with open(public_model, 'w+') as f:
f.write("from webapp.models._{} import {}".format(model_snake, model_camel))
f.write("from webapp.models._{}_acl import {}Acl\n".format(model_snake, model_camel))
f.write("from webapp.models._{} import {}\n".format(model_snake, model_camel))
def process_routes(view_model):
@ -220,8 +226,8 @@ def main():
process_routes(view_model)
view_model_name = view_model.get('name', '')
model_snake_name = camel_case_to_snake_case(view_model_name)
all_model_imports.append('from webapp.models.{} import {}'.format(
model_snake_name, view_model_name))
all_model_imports.append('from webapp.models.{s} import {c}, {c}Acl'.format(
s=model_snake_name, c=view_model_name))
process_html_templates(view_model)
all_route_imports.append('from webapp.routes.{} import *'.format(
model_snake_name