From 36b5ee74da34f72df7f89f620a575b45bc3242a4 Mon Sep 17 00:00:00 2001 From: Kevin Kirsche Date: Thu, 4 Aug 2022 16:29:07 -0400 Subject: [PATCH] refactor: add introductory type hints to flask-oso (#1606) * refactor: add introductory type hints to flask/oso * fix: failing test Co-authored-by: Sam Scott --- .../python/flask-oso/flask_oso/flask_oso.py | 65 ++++++++++++------- .../python/flask-oso/requirements-test.txt | 1 + .../python/flask-oso/tests/test_flask_oso.py | 2 +- 3 files changed, 44 insertions(+), 24 deletions(-) diff --git a/languages/python/flask-oso/flask_oso/flask_oso.py b/languages/python/flask-oso/flask_oso/flask_oso.py index 1ff1d64e92..b535be976d 100644 --- a/languages/python/flask-oso/flask_oso/flask_oso.py +++ b/languages/python/flask-oso/flask_oso/flask_oso.py @@ -1,7 +1,9 @@ -from flask import g, current_app, request, Request +from flask import g, current_app, request, Flask +from flask.wrappers import Request, Response from werkzeug.exceptions import Forbidden +from typing import Any, Callable, NoReturn, Optional -from oso import OsoError +from oso import Oso, OsoError from .context import _app_context @@ -34,11 +36,14 @@ class FlaskOso: Call `authorize(resource=flask.request)` before every request. """ - def __init__(self, oso=None, app=None): + _app: Optional[Flask] + _oso: Optional[Oso] + + def __init__(self, oso: Optional[Oso] = None, app: Optional[Flask] = None) -> None: self._app = app self._oso = None - def unauthorized(): + def unauthorized() -> NoReturn: raise Forbidden("Unauthorized") self._unauthorized_action = unauthorized @@ -57,7 +62,7 @@ def unauthorized(): # Initialization - def set_oso(self, oso): + def set_oso(self, oso: Oso) -> None: """Set the oso instance to use for authorization Must be called if ``oso`` is not provided to the constructor. @@ -68,7 +73,7 @@ def set_oso(self, oso): self._oso = oso self._oso.register_class(Request) - def init_app(self, app): + def init_app(self, app: Flask) -> None: """Initialize ``app`` for use with this instance of ``FlaskOso``. Must be called if ``app`` isn't provided to the constructor. @@ -76,7 +81,7 @@ def init_app(self, app): app.teardown_appcontext(self.teardown) app.before_request(self._provide_oso) - def set_get_actor(self, func): + def set_get_actor(self, func: Callable[[], Any]) -> None: """Provide a function that oso will use to get the current actor. :param func: A function to call with no parameters to get the actor if @@ -85,7 +90,7 @@ def set_get_actor(self, func): """ self._get_actor = func - def set_unauthorized_action(self, func): + def set_unauthorized_action(self, func: Callable[[], Any]) -> None: """Set a function that will be called to handle an authorization failure. The default behavior is to raise a Forbidden exception, returning a 403 @@ -98,7 +103,7 @@ def set_unauthorized_action(self, func): # Middleware-like functions that affect every request. - def require_authorization(self, app=None): + def require_authorization(self, app: Optional[Flask] = None) -> None: """Enforce authorization on every request to ``app``. :param app: The app to require authorization for. Can be omitted if @@ -112,11 +117,12 @@ def require_authorization(self, app=None): request. """ if app is None: - app = self.app - + app = self._app + if app is None: + raise OsoError("Cannot require authorization without Flask app object") app.after_request(self._require_authorization) - def perform_route_authorization(self, app=None): + def perform_route_authorization(self, app: Optional[Flask] = None) -> None: """Perform route authorization before every request. Route authorization will call :py:meth:`oso.Oso.is_allowed` with the @@ -128,13 +134,17 @@ def perform_route_authorization(self, app=None): constructor. """ if app is None: - app = self.app + app = self._app + if app is None: + raise OsoError( + "Cannot perform route authorization without Flask app object" + ) app.before_request(self._perform_route_authorization) # During request decorator or functions. - def skip_authorization(self, reason=None): + def skip_authorization(self, reason: Optional[str] = None) -> None: """Opt-out of authorization for the current request. Will prevent ``require_authorization`` from causing an error. @@ -143,7 +153,13 @@ def skip_authorization(self, reason=None): """ _authorize_called() - def authorize(self, resource, *, actor=None, action=None): + def authorize( + self, + resource: Any, + *, + actor: Optional[Any] = None, + action: Optional[str] = None + ) -> None: """Check whether the current request should be allowed. Calls :py:meth:`oso.Oso.is_allowed` to check authorization. If a request @@ -179,7 +195,10 @@ def authorize(self, resource, *, actor=None, action=None): # We use *is* here because == would actually need to get the request object. # We want to check that the resource is the proxy. if resource is request: - resource = request._get_current_object() + resource = request._get_current_object() # type: ignore + + if self.oso is None: + raise OsoError("Cannot perform authorization without oso instance") allowed = self.oso.is_allowed(actor, action, resource) _authorize_called() @@ -188,24 +207,24 @@ def authorize(self, resource, *, actor=None, action=None): self._unauthorized_action() @property - def app(self): + def app(self) -> Flask: return self._app or current_app @property - def oso(self): + def oso(self) -> Optional[Oso]: return self._oso @property - def current_actor(self): + def current_actor(self) -> Any: return self._get_actor() # Before / after - def _provide_oso(self): + def _provide_oso(self) -> None: top = _app_context() if not hasattr(top, "oso_flask_oso"): top.oso_flask_oso = self - def _perform_route_authorization(self): + def _perform_route_authorization(self) -> None: if not request.url_rule: # If request isn't going to match any route, just return and # let flask handle it the normal way. @@ -213,7 +232,7 @@ def _perform_route_authorization(self): self.authorize(resource=request) - def _require_authorization(self, response): + def _require_authorization(self, response: Response) -> Response: if not request.url_rule: # No rule matched this request # Skip requiring authorization. @@ -230,6 +249,6 @@ def teardown(self, exception): pass -def _authorize_called(): +def _authorize_called() -> None: """Mark current request as authorized.""" _app_context().oso_flask_authorize_called = True diff --git a/languages/python/flask-oso/requirements-test.txt b/languages/python/flask-oso/requirements-test.txt index c2845bffbe..8f1cc51dae 100644 --- a/languages/python/flask-oso/requirements-test.txt +++ b/languages/python/flask-oso/requirements-test.txt @@ -1 +1,2 @@ +types-setuptools==63.2.0 pytest==7.0.1 diff --git a/languages/python/flask-oso/tests/test_flask_oso.py b/languages/python/flask-oso/tests/test_flask_oso.py index dd57358825..a8ee1f20f7 100644 --- a/languages/python/flask-oso/tests/test_flask_oso.py +++ b/languages/python/flask-oso/tests/test_flask_oso.py @@ -48,7 +48,7 @@ def test_initialization_with_set(flask_app, oso, simple_policy, app_ctx, user): # Establish that an improperly initialized flask oso throws an exception. flask_oso = FlaskOso() flask_oso.set_get_actor(lambda: user) - with pytest.raises(AttributeError): + with pytest.raises(OsoError): flask_oso.authorize(action="read", resource="resource") # Works after set oso.