This is a follow-up to a previous article and will build on the ideas introduced therein - please check that out first!
Intro
Permissions! They sit at the deeply uncomfortable intersection of complex, business critical yet non-functional. Every organisation needs it and needs it done well, but will never be rewarded for it.
The previous post introduced some ideas around a repository pattern for a Python + relational database app, expressing access-based reads in terms of domain objects only.
So far, so neat.
…But what about writes?
Now things are really getting interesting! It’s the difference between a thought experiment for the curious and an implementable permissions system.
Such systems are really quite tricky to do well - where “good” would be a permissions system that is watertight yet flexible, intelligible yet unintrusive.
This is obviously an enormous challenge. As with most technical problems, this one benefits from repeatedly poking at the issue over a matter of days. Inspired from a conversation with a previous manager, I like to imagine them some extraordinarily elaborate slip knot - superficially very complicated, but reveal themselves as having a working end which, when pulled, allows the complexity to fall away.
The trick is to find the working end.
Hunting the working end
As an entrypoint to the problem, let’s focus on something concrete. Suppose that we need to apply the following rules:
- A user can see themselves
- A user can only update their own email
- A user can create a group
- A user belonging to a group can see other users in that group
- A user cannot edit a group they do not own
Our callers will look something as follows:
def update_group(db: DBSession, user: User, group_uid: str, *, name: str) -> UserGroup:
repo = Repository(db, UserGroup)
group = repo.update(
group_uid,
{"name": name},
guards=permissions(user, UserGroup, Operation.UPDATE),
)
db.commit()
return groupwhere you can see two new ideas from the previous repository pattern:
- The introduction of an
Operationthat denotes the action being taken by some user on some subject. - The introduction of a
Guardthat protects against write operations as a complement to a filter on read operations.
The operation is a fairly straightforward Enum value:
class Operation(StrEnum):
READ = "read"
CREATE = "create"
UPDATE = "update"The Guard is rather more interesting - especially its relationship to a Filter. Where a Filter acted on records in the database, the Guard must act on prospective database records, held in memory:
class Guard(Protocol):
"""Evaluable against an in-memory instance.
Usable to guard an INSERT, where there is no persisted row to attach a
``WHERE`` to.
"""
def is_satisfied_by(self, obj: BaseModel) -> bool: ...What’s interesting is that these are two different expressions of the same idea: a callable that forms the basis of a permissions decision. In functional programming jargon, we might call such functions a Predicate. So, we can generalise the original idea, applicable only for filtering, towards any operation that might be performed on a resource that we own.
So, where we had this before:
class EqualsFilter[T](Filter):
"""Filter for exact equality comparison.
Attributes:
field: The field name to filter on.
value: The value to compare for equality.
Example:
EqualsFilter(field="archived", value=False)
# Generates: WHERE archived = False
"""
value: T = Field(..., description="Value to compare for equality")
def apply(
self, stmt: Select[tuple[Any, ...]], entity: type[BaseModel]
) -> Select[tuple[Any, ...]]:
"""Apply equality filter to the statement."""
column = getattr(entity, self.field)
return stmt.where(column == self.value)We can now imagine this:
class Predicate(BaseSchema, ABC):
"""Base for all concrete predicates.
Carries the ``field`` attribute and declares the universal ``to_sql``
contract. It is the type used for nested fields (``list[Predicate]``,
``Predicate``) so Pydantic can construct nested predicate trees.
NOTE: there is intentionally no ``is_satisfied_by`` here — its absence is the
structural discriminator that keeps SQL-only predicates out of
``create(guards=...)``.
"""
field: str = Field(
..., description="Name of the model field the predicate refers to"
)
@abstractmethod
def to_sql(self, entity: type[BaseModel]) -> ColumnElement[bool]:
"""Render this predicate to a SQLAlchemy boolean expression."""
class Equals[T](Predicate):
"""Exact equality comparison.
Example:
Equals(field="archived", value=False) # archived = False
"""
value: T = Field(..., description="Value to compare for equality")
def to_sql(self, entity: type[BaseModel]) -> ColumnElement[bool]:
return getattr(entity, self.field) == self.value
def is_satisfied_by(self, obj: BaseModel) -> bool:
return bool(getattr(obj, self.field) == self.value)
def permissions(
user: User,
entity: type[BaseModel],
operation: Operation = Operation.READ,
) -> list[Filter] | list[Guard]:
...
if entity is User and operation is Operation.UPDATE:
# A user may only update their own row.
return [Equals(field="uid", value=user.uid)]
...Which is neat, as all our existing uses of Filters continue to function, calling the to_sql method, whilst the Guard gives us all we need to protect on both CREATE and UPDATE operations!
class Repository[T]:
def create(self, obj: T, guards: Sequence[Guard] = ()) -> T:
for guard in guards:
if not guard.is_satisfied_by(obj):
msg = (
f"{type(obj).__name__} instance rejected by write guard "
f"{type(guard).__name__}"
)
raise NotAuthorisedError(msg)
self.db.add(obj)
self.db.flush()
return obj(Un)Tying it all together
We started with the requirements that:
- A user can see themselves
- A user can only update their own email
- A user can create a group
- A user belonging to a group can see other users in that group
- A user cannot edit a group they do not own
We can now see that our permissions will look as follows:
def permissions(
user: User,
entity: type[BaseModel],
operation: Operation = Operation.READ,
) -> list[Filter] | list[Guard]:
is_me = Equals(field="uid", value=user.uid)
if entity is User:
if operation is Operation.READ:
return [
Or(
predicates=[
# (1) A user can see themselves...
is_me,
# (4) ...and anyone in a group they're also a member of.
Related(
field="user_groups",
inner=Related(field="members", inner=is_me),
),
]
)
]
if operation is Operation.UPDATE:
# (2) A user can only update their own row.
return [is_me]
if entity is UserGroup:
is_owner = Equals(field="owner_uid", value=user.uid)
if operation is Operation.CREATE:
# (3) A user can create a group — as its owner.
return [is_owner]
if operation is Operation.UPDATE:
# (5) A user cannot edit a group they do not own.
return [is_owner]
msg = f"No {operation} permission predicates defined for {entity.__name__}"
raise ValueError(msg)Cool! We have working write protections, all whilst adhering to a clean architecture that remains entirely unit testable!
System Evolution
We just demonstrated how the addition of a new requirement led us to identifying the need for a rethink on our system prior to its next iteration. We have a word for this: a refactor!
Refactors are the means by which we avoid being this poor guy:
A lot has been said on refactors. I don’t have much novel to contribute here, just a signpost to existing ideas.
The most succinct (and entertaining) take on this comes from Carson Gross’s Grug-Brained Developer. In short:
- In short, your understanding of a system is always best towards the end of a project. So it’s at this point that your code is best amended to reflect that understanding.
- Don’t go far out from shore during your refactor - keep the whole system working whilst you refactor.
And, if you can smuggle in your refactors under the banner of product intiatives, you’ll be a hero to your fellow Engineers.