feat: permission permissions
All checks were successful
ci / lint_and_test (push) Successful in 16s

Orgs can only grant permissions to groups that they themselves have been granted access to.

Super admin bypasses not added, flagged as todos.
This commit is contained in:
Chris Milne 2026-06-16 13:51:31 +01:00
parent 0a867c9c90
commit 662b9c8e26
6 changed files with 71 additions and 4 deletions

View file

@ -0,0 +1,38 @@
"""org perm perms join table
Revision ID: 85edbf9a176c
Revises: 98e20aae555c
Create Date: 2026-06-16 13:31:57.427953
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '85edbf9a176c'
down_revision: Union[str, Sequence[str], None] = '98e20aae555c'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('org_permissions',
sa.Column('org_id', sa.Integer(), nullable=False),
sa.Column('permission_id', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['org_id'], ['organisation.id'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['permission_id'], ['permission.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('org_id', 'permission_id')
)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('org_permissions')
# ### end Alembic commands ###

View file

@ -52,6 +52,10 @@ class Permission(Base):
"Group", secondary="group_permissions", back_populates="permission_rel" "Group", secondary="group_permissions", back_populates="permission_rel"
) )
org_rel = relationship(
"Organisation", secondary="org_permissions", back_populates="permission_rel"
)
class Group(Base): class Group(Base):
__tablename__ = "group" __tablename__ = "group"
@ -95,3 +99,13 @@ class UserGroups(Base):
group_id = Column( group_id = Column(
Integer, ForeignKey("group.id", ondelete="CASCADE"), primary_key=True Integer, ForeignKey("group.id", ondelete="CASCADE"), primary_key=True
) )
class OrgPermissions(Base):
__tablename__ = "org_permissions"
org_id = Column(
Integer, ForeignKey("organisation.id", ondelete="CASCADE"), primary_key=True
)
permission_id = Column(
Integer, ForeignKey("permission.id", ondelete="CASCADE"), primary_key=True
)

View file

@ -325,6 +325,9 @@ async def add_group_permission(
if perm_model in group_model.permission_rel: if perm_model in group_model.permission_rel:
raise ConflictException("Group already has this permission") raise ConflictException("Group already has this permission")
if perm_model not in org_model.permission_rel: # TODO: and not su
raise ForbiddenException("You cannot grant this permission")
group_model.permission_rel.append(perm_model) group_model.permission_rel.append(perm_model)
db.flush() db.flush()
@ -471,8 +474,10 @@ async def get_permissions(
""" """
Returns a full list of permissions. Returns a full list of permissions.
""" """
permission_models = db.query(Perm).all() # TODO: if su:
# permission_models = db.query(Perm).all()
# else
permission_models = db.query(Perm).filter(Perm.org_rel.any(id=org_model.id)).all()
return {"permissions": permission_models} return {"permissions": permission_models}
@ -566,6 +571,9 @@ async def post_permissions(
if not (request_model.action is None or request_model.action == ""): if not (request_model.action is None or request_model.action == ""):
permission_query = permission_query.filter(Perm.action == request_model.action) permission_query = permission_query.filter(Perm.action == request_model.action)
# TODO: if not su:
permission_query = permission_query.filter(Perm.org_rel.any(id=org_model.id))
permission_models = permission_query.all() permission_models = permission_query.all()
return {"permissions": permission_models} return {"permissions": permission_models}

View file

@ -55,6 +55,10 @@ class Organisation(Base):
"Contact", foreign_keys="Organisation.owner_contact_id" "Contact", foreign_keys="Organisation.owner_contact_id"
) )
permission_rel = relationship(
"Permission", secondary="org_permissions", back_populates="org_rel"
)
class OrgUsers(Base): class OrgUsers(Base):
__tablename__ = "orgusers" __tablename__ = "orgusers"

View file

@ -10,7 +10,7 @@ from src.user.models import User
from src.service.models import Service from src.service.models import Service
from src.organisation.models import Organisation as Org, OrgUsers from src.organisation.models import Organisation as Org, OrgUsers
from src.contact.models import Contact from src.contact.models import Contact
from src.iam.models import Group, Permission from src.iam.models import Group, Permission, OrgPermissions
from src.auth.service import get_current_user, get_dev_user from src.auth.service import get_current_user, get_dev_user
from src.auth.dependencies import empty_su_list, get_super_admin_list, testing_su_list from src.auth.dependencies import empty_su_list, get_super_admin_list, testing_su_list
from src.main import app # inited FastAPI app from src.main import app # inited FastAPI app
@ -163,6 +163,9 @@ def _seed(db):
db.add(Service(name="Test Service", api_key="123456789")) db.add(Service(name="Test Service", api_key="123456789"))
db.add(Permission(service_id=1, resource="test_resource", action="read")) db.add(Permission(service_id=1, resource="test_resource", action="read"))
db.add(Permission(service_id=1, resource="test_resource", action="move")) db.add(Permission(service_id=1, resource="test_resource", action="move"))
db.add(Permission(service_id=1, resource="test_resource", action="delete"))
db.add(OrgPermissions(org_id=1, permission_id=1))
db.add(OrgPermissions(org_id=1, permission_id=2))
db.add(Group(name="Org One Group", org_id=1)) db.add(Group(name="Org One Group", org_id=1))
db.add(Group(name="Org Two Group", org_id=2)) db.add(Group(name="Org Two Group", org_id=2))
db.add(Group(name="Org One Group Two", org_id=1)) db.add(Group(name="Org One Group Two", org_id=1))

View file

@ -437,7 +437,7 @@ async def test_post_perm_success(default_client: AsyncClient):
assert "permission" in data assert "permission" in data
assert isinstance(data["permission"], dict) assert isinstance(data["permission"], dict)
assert data["permission"]["id"] == 3 assert data["permission"]["id"] == 4
assert data["permission"]["service_name"] == "Test Service" assert data["permission"]["service_name"] == "Test Service"
assert data["permission"]["resource"] == "test_resource" assert data["permission"]["resource"] == "test_resource"
assert data["permission"]["action"] == "create" assert data["permission"]["action"] == "create"