Merge pull request #55 from kaiyou/feat-refactor-permissions
Refactor the access control codemaster
kaiyou
8 years ago
committed by
GitHub
commit
cbc6bb5dd6
@ -0,0 +1,43 @@
|
||||
from freeposte import app
|
||||
|
||||
import sys
|
||||
import tabulate
|
||||
|
||||
|
||||
# Known endpoints without permissions
|
||||
known_missing_permissions = [
|
||||
"index",
|
||||
"static", "bootstrap.static",
|
||||
"admin.static", "admin.login"
|
||||
]
|
||||
|
||||
|
||||
# Compute the permission table
|
||||
missing_permissions = []
|
||||
permissions = {}
|
||||
for endpoint, function in app.view_functions.items():
|
||||
audit = function.__dict__.get("_audit_permissions")
|
||||
if audit:
|
||||
handler, args = audit
|
||||
if args:
|
||||
model = args[0].__name__
|
||||
key = args[1]
|
||||
else:
|
||||
model = key = None
|
||||
permissions[endpoint] = [endpoint, handler.__name__, model, key]
|
||||
elif endpoint not in known_missing_permissions:
|
||||
missing_permissions.append(endpoint)
|
||||
|
||||
|
||||
# Fail if any endpoint is missing a permission check
|
||||
if missing_permissions:
|
||||
print("The following endpoints are missing permission checks:")
|
||||
print(missing_permissions.join(","))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
# Display the permissions table
|
||||
print(tabulate.tabulate([
|
||||
[route, *permissions[route.endpoint]]
|
||||
for route in app.url_map.iter_rules() if route.endpoint in permissions
|
||||
]))
|
||||
@ -0,0 +1,114 @@
|
||||
from freeposte.admin import db, models, forms
|
||||
|
||||
import flask
|
||||
import flask_login
|
||||
import functools
|
||||
|
||||
|
||||
def permissions_wrapper(handler):
|
||||
""" Decorator that produces a decorator for checking permissions.
|
||||
"""
|
||||
def callback(function, args, kwargs, dargs, dkwargs):
|
||||
authorized = handler(args, kwargs, *dargs, **dkwargs)
|
||||
if not authorized:
|
||||
flask.abort(403)
|
||||
elif type(authorized) is int:
|
||||
flask.abort(authorized)
|
||||
else:
|
||||
return function(*args, **kwargs)
|
||||
# If the handler has no argument, declare a
|
||||
# simple decorator, otherwise declare a nested decorator
|
||||
# There are at least two mandatory arguments
|
||||
if handler.__code__.co_argcount > 2:
|
||||
def decorator(*dargs, **dkwargs):
|
||||
def inner(function):
|
||||
@functools.wraps(function)
|
||||
def wrapper(*args, **kwargs):
|
||||
return callback(function, args, kwargs, dargs, dkwargs)
|
||||
wrapper._audit_permissions = handler, dargs
|
||||
return flask_login.login_required(wrapper)
|
||||
return inner
|
||||
else:
|
||||
def decorator(function):
|
||||
@functools.wraps(function)
|
||||
def wrapper(*args, **kwargs):
|
||||
return callback(function, args, kwargs, (), {})
|
||||
wrapper._audit_permissions = handler, []
|
||||
return flask_login.login_required(wrapper)
|
||||
return decorator
|
||||
|
||||
|
||||
@permissions_wrapper
|
||||
def global_admin(args, kwargs):
|
||||
""" The view is only available to global administrators.
|
||||
"""
|
||||
return flask_login.current_user.global_admin
|
||||
|
||||
|
||||
@permissions_wrapper
|
||||
def domain_admin(args, kwargs, model, key):
|
||||
""" The view is only available to specific domain admins.
|
||||
Global admins will still be able to access the resource.
|
||||
|
||||
A model and key must be provided. The model will be queries
|
||||
based on the query parameter named after the key. The model may
|
||||
either be Domain or an Email subclass (or any class with a
|
||||
``domain`` attribute which stores a related Domain instance).
|
||||
"""
|
||||
obj = model.query.get(kwargs[key])
|
||||
if not obj:
|
||||
flask.abort(404)
|
||||
else:
|
||||
domain = obj if type(obj) is models.Domain else obj.domain
|
||||
return domain in flask_login.current_user.get_managed_domains()
|
||||
|
||||
|
||||
@permissions_wrapper
|
||||
def owner(args, kwargs, model, key):
|
||||
""" The view is only available to the resource owner or manager.
|
||||
|
||||
A model and key must be provided. The model will be queries
|
||||
based on the query parameter named after the key. The model may
|
||||
either be User or any model with a ``user`` attribute storing
|
||||
a user instance (like Fetch).
|
||||
|
||||
If the query parameter is empty and the model is User, then
|
||||
the resource being accessed is supposed to be the current
|
||||
logged in user and access is obviously authorized.
|
||||
"""
|
||||
if kwargs[key] is None and model == models.User:
|
||||
return True
|
||||
obj = model.query.get(kwargs[key])
|
||||
if not obj:
|
||||
flask.abort(404)
|
||||
else:
|
||||
user = obj if type(obj) is models.User else obj.user
|
||||
return (
|
||||
user.email == flask_login.current_user.email
|
||||
or user.domain in flask_login.current_user.get_managed_domains()
|
||||
)
|
||||
|
||||
|
||||
@permissions_wrapper
|
||||
def authenticated(args, kwargs):
|
||||
""" The view is only available to logged in users.
|
||||
"""
|
||||
return True
|
||||
|
||||
|
||||
|
||||
def confirmation_required(action):
|
||||
""" View decorator that asks for a confirmation first.
|
||||
"""
|
||||
def inner(function):
|
||||
@functools.wraps(function)
|
||||
def wrapper(*args, **kwargs):
|
||||
form = forms.ConfirmationForm()
|
||||
if form.validate_on_submit():
|
||||
return function(*args, **kwargs)
|
||||
return flask.render_template(
|
||||
"confirm.html", action=action.format(*args, **kwargs),
|
||||
form=form
|
||||
)
|
||||
return wrapper
|
||||
return inner
|
||||
@ -1,69 +0,0 @@
|
||||
from freeposte.admin import models, forms
|
||||
|
||||
import flask
|
||||
import flask_login
|
||||
import functools
|
||||
|
||||
|
||||
def confirmation_required(action):
|
||||
""" View decorator that asks for a confirmation first.
|
||||
"""
|
||||
def inner(function):
|
||||
@functools.wraps(function)
|
||||
def wrapper(*args, **kwargs):
|
||||
form = forms.ConfirmationForm()
|
||||
if form.validate_on_submit():
|
||||
return function(*args, **kwargs)
|
||||
return flask.render_template(
|
||||
"confirm.html", action=action.format(*args, **kwargs),
|
||||
form=form
|
||||
)
|
||||
return wrapper
|
||||
return inner
|
||||
|
||||
|
||||
def get_domain_admin(domain_name):
|
||||
domain = models.Domain.query.get(domain_name)
|
||||
if not domain:
|
||||
flask.abort(404)
|
||||
if not domain in flask_login.current_user.get_managed_domains():
|
||||
flask.abort(403)
|
||||
return domain
|
||||
|
||||
|
||||
def require_global_admin():
|
||||
if not flask_login.current_user.global_admin:
|
||||
flask.abort(403)
|
||||
|
||||
|
||||
def get_user(user_email, admin=False):
|
||||
if user_email is None:
|
||||
user_email = flask_login.current_user.email
|
||||
user = models.User.query.get(user_email)
|
||||
if not user:
|
||||
flask.abort(404)
|
||||
if not user.domain in flask_login.current_user.get_managed_domains():
|
||||
if admin:
|
||||
flask.abort(403)
|
||||
elif not user.email == flask_login.current_user.email:
|
||||
flask.abort(403)
|
||||
return user
|
||||
|
||||
|
||||
def get_alias(alias):
|
||||
alias = models.Alias.query.get(alias)
|
||||
if not alias:
|
||||
flask.abort(404)
|
||||
if not alias.domain in flask_login.current_user.get_managed_domains():
|
||||
return 403
|
||||
return alias
|
||||
|
||||
|
||||
def get_fetch(fetch_id):
|
||||
fetch = models.Fetch.query.get(fetch_id)
|
||||
if not fetch:
|
||||
flask.abort(404)
|
||||
if not fetch.user.domain in flask_login.current_user.get_managed_domains():
|
||||
if not fetch.user.email == flask_login.current_user.email:
|
||||
flask.abort(403)
|
||||
return fetch
|
||||
Loading…
Reference in New Issue