From 6178ef662c9a8f62bddd48125b9fd5afc4b5476f Mon Sep 17 00:00:00 2001 From: Ferenc Schulcz <schulcz.ferenc@gmail.com> Date: Fri, 11 Oct 2024 18:01:30 +0200 Subject: [PATCH] Better registration scheme --- router.py | 105 +++++++++++++++++++++++++++++++--------------------- services.py | 10 +++-- 2 files changed, 69 insertions(+), 46 deletions(-) diff --git a/router.py b/router.py index 31c1cb7..4f007a2 100644 --- a/router.py +++ b/router.py @@ -30,31 +30,37 @@ def make_session_permanent(): # Set up endpoint database ############################################################### -service_endpoints = dict() -postable_service_endpoints = dict() - - -def plugin_add_endpoint(name: str, handler): - service_endpoints[name] = handler -def plugin_add_postable_endpoint(name: str, handler): - postable_service_endpoints[name] = handler - - -service_menu = [] - - -def plugin_add_menu(name: str, text: str, endpoint: str): - service_menu.append((name, text, endpoint)) - - -## register service plugins -## vm_service.register(plugin_add_endpoint) -#modules = glob.glob(join(dirname(__file__)+'/service_plugins', "*.py")) -#modules = [basename(f)[:-3] for f in modules if isfile(f) -# and not f.endswith('__init__.py')] -#for module in modules: -# getattr(sys.modules['service_plugins.'+module], -# "register")(plugin_add_endpoint, plugin_add_postable_endpoint, plugin_add_menu) +public_service_endpoints = dict() +public_service_menu = [] +authorized_service_endpoints = dict() +authorized_service_menu = [] + +def plugin_add_endpoint(endpoint_id: str, handler, permission_name: str, method='GET', menutext=None): + """Add a dynamic endpoint.""" + """ id: used in the url (like /something)""" + """ handler: pointer to handler function""" + """ servicename: service identifier for authorization. If None, it is a public (and public-only) endpoint.""" + """ method: HTTP method this endpoint can handle. If you plan to handle multiple endpoints, call this function multiple times. (Make sure to only add a menu once.)""" + """ menutext: Text to appear in the menu. If None, no menu item is added.""" + + if permission_name == None: + # Public endpoint + # add endpoint + if not method in public_service_endpoints: + public_service_endpoints[method] = dict() + public_service_endpoints[method][endpoint_id] = handler + # add to menu + if not menutext == None: + public_service_menu.append((menutext, endpoint_id)) + else: + # Authorized endpoint + # add endpoint + if not method in authorized_service_endpoints: + authorized_service_endpoints[method] = dict() + authorized_service_endpoints[method][endpoint_id] = (permission_name, handler) + # add to menu + if not menutext == None: + authorized_service_menu.append((menutext, endpoint_id, permission_name)) # Import and register plugins @@ -75,7 +81,7 @@ for plugin_dir in os.listdir(join(dirname(__file__)+'/plugins')): # calling register() if hasattr(module, "register"): - module.register(plugin_add_endpoint, plugin_add_postable_endpoint, plugin_add_menu) + module.register(plugin_add_endpoint) else: print(' ' + plugin + ' has no register() function.') @@ -284,29 +290,44 @@ def get_500(error, **kwargs): @ app.context_processor def services_processor(): def get_service_menus(username: str): - return services.get_service_menus(username, service_menu) + return services.get_service_menus(username, public_service_menu, authorized_service_menu) return dict(get_service_menus=get_service_menus, getMessages=db.getMessages, hasMessages=db.hasMessages) -@ app.route('/<servicename>') +def get_public_endpoint(servicename: str, method: str): + if not method in public_service_endpoints: + return None + if not servicename in public_service_endpoints[method]: + return None + return public_service_endpoints[method][servicename] + +def get_authorized_endpoint(servicename: str, method: str): + if not method in authorized_service_endpoints: + return None + if not servicename in authorized_service_endpoints[method]: + return None + return authorized_service_endpoints[method][servicename] + +@ app.route('/<servicename>', methods=['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH']) def service(**kwargs): servicename = kwargs['servicename'] - if not servicename in service_endpoints.keys(): - return get_404(None) - handler = service_endpoints[servicename] - if not handler is None: + if 'username' not in session.keys(): + # Search for public endpoints first. If no such endpoint is found, check is authorized endpoint exists. If so, redirect to login, else send 404. + handler = get_public_endpoint(servicename, request.method) + if handler == None: + (permission_name, handler) = get_authorized_endpoint(servicename, request.method) + if handler == None: + return get_404(None) + else: + return redirect(url_for('login', next=servicename)) return handler(session=session, request=request, rqtools=RequestTools()) - return get_404(None) - -@ app.route('/<servicename>', methods=['POST']) -def postservice(**kwargs): - servicename = kwargs['servicename'] - if not servicename in postable_service_endpoints.keys(): - return get_404(None) - handler = postable_service_endpoints[servicename] - if not handler is None: + else: + (permission_name, handler) = get_authorized_endpoint(servicename, request.method) + if handler == None: + return get_404(None) + if not services.authorize_user(session['username'], permission_name): + return get_403(None) return handler(session=session, request=request, rqtools=RequestTools()) - return get_404(None) @ app.route('/page/<pagename>') diff --git a/services.py b/services.py index 9f7489b..2d19bf4 100644 --- a/services.py +++ b/services.py @@ -2,17 +2,19 @@ import pymongo import db -def get_service_menus(username: str, service_menu: dict): +def get_service_menus(username: str, public_service_menu: dict, authorized_service_menu: dict): ret = {} if username is None: + for (text, service) in public_service_menu: + ret[text] = service return ret x = db.users.find_one(filter={'username': username}) if x is None: return ret for service_name in x['services']: - for s, text, endpoint in service_menu: - if s == service_name: - ret[text] = endpoint + for (text, service, permission_name) in authorized_service_menu: + if permission_name == service_name: + ret[text] = service return ret -- GitLab