Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions docs/content/any/project/changelogs/NEXT.md
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,13 @@ oso.query('f(x)', { bindings });
- Thanks to [`@Kn99HN`](https://github.com/Kn99HN) for adding the
`acceptExpression` query flag to the Node.js lib!

## `sqlalchemy-oso` `NEW_VERSION`

### Other bugs & improvements

- `scoped_session` now correctly handles a `get_checked_permission` callback that
returns `None`.

## `RELEASED_PACKAGE_1` NEW_VERSION

### LANGUAGE (e.g., 'Core' or 'Python' or 'Node.js')
Expand Down
4 changes: 2 additions & 2 deletions languages/python/sqlalchemy-oso/requirements-dev.txt
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
black==21.5b0
black~=21.12b
flake8==3.9.2
mypy==0.812
sqlalchemy-stubs==0.4
tox==3.23.1
tox==3.23.1
5 changes: 3 additions & 2 deletions languages/python/sqlalchemy-oso/sqlalchemy_oso/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,8 +176,9 @@ def scoped_session(
scopefunc = scopefunc or (lambda: None)

def _scopefunc():
checked_permissions = frozenset(get_checked_permissions().items())
return (get_oso(), checked_permissions, get_user(), scopefunc())
perms = get_checked_permissions()
perms = frozenset() if perms is None else frozenset(perms.items())
return (get_oso(), perms, get_user(), scopefunc())

factory = authorized_sessionmaker(
get_oso, get_user, get_checked_permissions, **kwargs
Expand Down
11 changes: 11 additions & 0 deletions languages/python/sqlalchemy-oso/tests/test_sqlalchemy.py
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,17 @@ def test_authorized_session_relationship(engine, oso, fixture_data):
assert post_7.created_by is None


def test_scoped_session_with_no_checked_permissions(engine, oso, fixture_data):
# the policy denies all requests
oso.load_str("allow(_, _, _) if false;")
# but passing None skips authorization
session = scoped_session(lambda: oso, lambda: "user", lambda: None)
session.configure(bind=engine)
posts = session.query(Post)
# check that any posts are allowed
assert posts.count()


def test_scoped_session_relationship(engine, oso, fixture_data):
oso.load_str(
"""allow("user", "read", post: Post) if post.id = 1;
Expand Down