sqlalchemy_oso.flask.AuthorizedSQLAlchemy docstring example fails
#1572
Comments
kkirsche
changed the title
sqlalchemy_oso.flask.AuthorizedSQLAlchemy docstring example fails
|
Hi @kkirsche - I'm taking a look at this. We definitely have examples for using Oso with SQLAlchemy and Flask together, but I'm not sure if we have examples of using Oso with |
|
Hi @kkirsche, it's an API that hasn't gotten a lot of love recently. This is the only instance I can find across our examples, but it's using a very old version of For a more recent example of Flask best practices w/ Oso, I would look at the "flask-sqlalchemy" backend of GitClub. |
|
Thanks, @gj, I appreciate you looking. Sadly the I'll probably monkey patch my own version to work with |
|
Didn't mean to leave you out @killpack — appreciate your time and assistance as well |
|
Yeah definitely not taken as a knock @kkirsche — it's valid, appreciated feedback |
|
Hey Folks, any news on this? The docs seem kinda confusing, since none of the examples have been touched for some time. It seems that there are "multiple" approaches to get oso working with flask/sqlalchemy/graphene but i find it quite confusing to narrow "a best" solution down. Im quite confused on the "how to setup |
|
Hey 👋 I will try to grab some of the code I ended up using and share it here to get you started. Ultimately, I ended up going with I've only run into one minor bug the other day with our authorization policy where for some reason ReBAC seems to not be applied properly for two people and it's been really hard to debug as most of the tooling was only given to oso cloud, but otherwise the experience has been good. I should be able to get you that code this afternoon as I'm in meetings most of the morning. |
|
OK, here is some of the code I'm using to set things up with Oso in a Flask application. For context, my (simplified) app structure is: I'll put the filename at the top of the code blocks for clarity. I'm using Python 3.11+ exclusively, so you may need to switch things up for your supported versions. This next file is the most important, as it contains the session decorator which injects the checked permissions # src/ourapp/authorization/authorization_initialization.py
from collections.abc import Callable, Mapping
from contextlib import suppress
from functools import partial, wraps
from importlib.resources import files
from typing import ParamSpec, TypeVar, cast
from flask import Flask, current_app, g
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeMeta, Session
from sqlalchemy_oso import SQLAlchemyOso
from sqlalchemy_oso.session import scoped_session
from ourapp.errors.libraries import ForbiddenError, NotFoundError, Unauthorized
from ourapp.models import Base, User
from ourapp.settings import settings
from ourapp.utilities import get_current_user
try:
from greenlet import getcurrent as scopefunc
except ImportError:
from threading import get_ident as scopefunc
P = ParamSpec("P")
RT = TypeVar("RT")
Permissions = Mapping[type[DeclarativeMeta] | type[Base], str]
def init_oso(app: Flask, distribution_name: str) -> None:
"""Initialize the Oso authorization system.
This is required to perform all downstream data filtering and authorization
decisions.
Args:
app: The Flask application to protect.
distribution_name (str): The distribution name where the authorization policy
is distributed in.
"""
oso = SQLAlchemyOso(sqlalchemy_base=Base)
oso.forbidden_error = ForbiddenError
oso.not_found_error = NotFoundError
# https://importlib-resources.readthedocs.io/en/latest/migration.html#pkg-resources-resource-filename # noqa: E501
absolute_policies = files(package=distribution_name)
oso.load_files(
filenames=[
str(absolute_policies / relative_policy_file_path)
for relative_policy_file_path in settings.api.authorization.policy_files
]
)
app.oso = oso # type: ignore[attr-defined]
app.authorized_sessionmaker = partial( # type: ignore[attr-defined]
scoped_session,
get_oso=get_oso,
get_user=get_user,
scopefunc=scopefunc,
)
def get_oso() -> SQLAlchemyOso:
"""Retrieve the SQLAlchemyOso instance from the protected Flask application.
Returns:
The SQLAlchemyOso instance. Used by the authorized sessionmaker to automate
applying the data filtering according to the authorization policy.
"""
return cast(SQLAlchemyOso, current_app.oso) # type: ignore[attr-defined]
def get_user() -> User:
"""Retrieve the authenticated user.
Returns:
The authenticated user. Used by the authorized sessionmaker to automate
applying the data filtering according to the authorization policy.
"""
return get_current_user()
def session(
checked_permissions: Permissions | None,
) -> Callable[[Callable[P, RT]], Callable[P, RT]]:
"""Generate the authorized session with the provided permissions to check.
Args:
checked_permissions (Permissions | None): The permissions to enforce.
Reference:
https://github.com/osohq/gitclub/blob/main/backends/flask-sqlalchemy-oso/app/routes/helpers.py#L13
""" # noqa: E501
def decorator(func: Callable[P, RT]) -> Callable[P, RT]:
"""The decorator that will be used once the outer session function is provided.
The purpose of this is to capture the mapping of permissions.
"""
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> RT:
"""The function to call before executing the route's function."""
g.session = cast(
Session,
current_app.authorized_sessionmaker( # type: ignore[attr-defined]
bind=cast(SQLAlchemy, current_app.extensions["sqlalchemy"]).engine,
get_checked_permissions=lambda: checked_permissions,
)(),
)
return func(*args, **kwargs)
return wrapper
return decoratorYou'd then use the session decorator in a way similar to this: @version_one_blueprint.get("/example/datacenters/locations")
@session(checked_permissions={DataCenterLocation: "read"})
@IAM.login_required
def get_list_of_locations() -> Response:
.... the code to get a list of locations...This takes care of our initialization work and is called from our app set up code. It mutates app, so it doesn't bother returning anything. The We needed these for our own debugging work: So that's here: # src/ourapp/authorization/dump.py
from typing import TypedDict
from flask import current_app
from ourapp.authorization.utilities import authorized_roles
from ourapp.models import (
...our protected models
)
from ourapp.utilities import get_current_user
resources = {
... our protected models
}
class Roles(TypedDict):
"""The structures of roles.
Attributes:
roles (list[str]): The list of Oso role names,
such as "maintainer" and "reader".
"""
roles: list[str]
class Actions(TypedDict):
"""The structures of actions.
Attributes:
actions (list[str]): The list of Oso allowed actions,
such as "read" and "write".
"""
actions: list[str]
class RolesAndPermissions(Roles, Actions):
"""The structures of actions.
Attributes:
actions (list[str]): The list of Oso allowed actions,
such as "read" and "write".
roles (list[str]): The list of Oso role names,
such as "maintainer" and "reader".
"""
pass
def dump_permissions_for_authenticated_user() -> dict[str, Actions]:
"""Returns the Oso authorized actions (permissions) for the current user.
The current user is the user as determined by their JWT Bearer token.
An action may be "read", "write", "update", "delete", etc.
Returns:
dict[str, Actions]: A mapping of resource to allowed actions.
"""
return {
resource.__name__: {
"actions": list(
current_app.oso.authorized_actions( # type: ignore[attr-defined]
get_current_user(), resource
)
)
}
for resource in resources
}
def dump_roles_for_authenticated_user() -> dict[str, Roles]:
"""Returns the Oso authorized roles for the current user.
The current user is the user as determined by their JWT Bearer token.
A role may be "maintainer", "reader", "creator", "owner", etc.
Returns:
dict[str, Roles]: A mapping of resource to roles.
"""
return {
resource.__name__: {
"roles": list(authorized_roles(get_current_user(), resource))
}
for resource in resources
}
def dump_all_for_authenticated_user(user: User) -> dict[str, RolesAndPermissions]:
"""Returns the Oso authorized actions and roles for the current user.
The current user is the user as determined by their JWT Bearer token.
A role may be "maintainer", "reader", "creator", "owner", etc.
An action may be "read", "write", "update", "delete", etc.
Returns:
dict[str, RolesAndPermissions]: A mapping of resource to roles and actions.
"""
return {
resource.__name__: {
"actions": list(
current_app.oso.authorized_actions( # type: ignore[attr-defined]
user, resource
)
),
"roles": list(authorized_roles(user, resource)),
}
for resource in resources
}Which brings us to: # src/ourapp/authorization/utilities.py
from typing import cast
from flask import current_app
from polar.variable import Variable
from sqlalchemy.orm import DeclarativeMeta
from sqlalchemy_oso import SQLAlchemyOso
from ourapp.models import User
from ourapp.utilities import get_current_user
def authorized_roles(
actor: User, resource: DeclarativeMeta | type[DeclarativeMeta]
) -> set[str]:
"""Returns a list of authorized roles
Source:
https://github.com/osohq/oso/issues/1427
Args:
actor: (User): The user / actor who is performing the action(s).
resource (Base): The SQLAlchemy resource being authorized.
Raises:
ValueError: When there is no authenticated user.
ValueError: When there is no oso instance.
Returns:
set[str]: The list of role names found matching the user in the policy.
"""
role_results = current_app.oso.query_rule( # type: ignore[attr-defined]
"has_role", actor, Variable("role"), resource
)
return {result.get("bindings").get("role") for result in role_results}
def authorize_current_user(
action: str, resource: object, check_read: bool = True
) -> None:
"""A utility method to check if the currently authenticated user is authorized to
perform the requested action on the resource.
Args:
action: The action to authorize the user for.
resource: The resource, most commonly a database model instance, to authorize
the action on. Note that `object` is used as the type hint instead of `Any`
as this method supports any registered object. Per the following discussion
with the Mypy typing team, this aligns with the meaning of the type more
accurately than Any, which is intended only to silence errors.
https://github.com/python/typeshed/pull/8469#issuecomment-1203109282
check_read (optional): Should the authorization ensure the user also has the
read permission for the resource? This should be disabled in cases where a
standalone permission exists that is separate from reading the actual data,
such as aggregations or metrics. Defaults to True.
"""
cast(SQLAlchemyOso, current_app.oso).authorize( # type: ignore[attr-defined]
actor=get_current_user(),
action=action,
resource=resource,
check_read=check_read,
)This also references the get current user from our utilities: """The typed_retrievers module is used to return objects from Flask's global context
that are appropriately typed rather than `Any`. This helps avoid the use of cast
throughout the application and makes it easier to swap out usage of these values
in the future if required.
"""
# src/ourapp/utilities/typed_retrievers.py
from flask import g
from sqlalchemy.orm import Session
from ourapp.errors.libraries import Unauthorized
from ourapp.models import User
def get_current_user() -> User:
try:
match g.current_user:
case User():
return g.current_user
case None:
raise Unauthorized("Please login to perform this action.")
case type():
name = g.current_user.__name__
case object():
name = g.current_user.__class__.__name__
case _:
# this should be impossible to reach because object is any type in
# python for more details, see:
# https://mypy.readthedocs.io/en/stable/dynamic_typing.html#any-vs-object
name = repr(g.current_user)
raise ValueError(
f"Expected g.current_user to be a '{User.__name__}' instance, received "
+ f"'{name}'"
)
except AttributeError as ae:
raise Unauthorized("Please login to perform this action.") from ae
def get_authorized_session() -> Session:
if isinstance(g.session, Session):
return g.session
try:
raise ValueError(
f"Expected g.session to be a '{Session.__name__}' instance, "
+ f"received '{g.session.__name__}'."
)
except AttributeError as ae:
raise ValueError(
f"Expected g.session to be a '{Session.__name__}' instance, "
+ f"received '{type(g.session)}'."
) from aeFinally, while our errors are rather benign, I figured I'd share them anyway: # src/ourapp/errors/libraries/oso.py
from oso.exceptions import ForbiddenError as OsoForbiddenError
from oso.exceptions import NotFoundError as OsoNotFoundError
from werkzeug.exceptions import Forbidden as WerkzeugForbidden
from werkzeug.exceptions import HTTPException
from werkzeug.exceptions import NotFound as WerkzeugNotFound
from werkzeug.exceptions import Unauthorized as WekzeugUnauthorized
from ourapp.errors.ourapp import OurAppError
class WerkzeugError(OurAppError, HTTPException):
"""A Werkzeug HTTP error for use in our app."""
pass
class ForbiddenError(WerkzeugError, WerkzeugForbidden, OsoForbiddenError):
"""A Forbidden error for use in our app. Wraps upstream forbidden types."""
pass
class NotFoundError(WerkzeugError, WerkzeugNotFound, OsoNotFoundError):
"""A Not Found error for use in our app. Wraps upstream not found types."""
pass
class Unauthorized(WerkzeugError, WekzeugUnauthorized):
"""An Unauthorized error for use in our app. Wraps upstream unauthorized types."""
passSorry for the wall of code, but I hope that helps with getting you started. |
When using
AuthorizedSQLAlchemyin a flask application, if usingget_checked_permissionsas the example shows withflask.request.methodas the action to check, Flask throws:Docstring:
implies that this would be created for use throughout the application when the application starts to take advantage of connection pools
Do you have any functional examples using flask_sqlalchemy with get_checked_permissions?
The text was updated successfully, but these errors were encountered: