agilentics / boiler 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
"""Who is asking, and what they are allowed to ask for.

`Principal` is the seam that multi-tenancy and authentication plug into. It is
resolved once by a façade (from a session cookie, an API token, an OIDC callback)
and threaded through the core, so there is one answer to "may this caller do
this" regardless of which door the request came through.

Nothing here is HTTP-flavoured: the core raises domain errors and each façade
maps them to whatever its transport calls a failure (see endpoints/crud.STATUS).
"""
from dataclasses import dataclass, field


class CoreError(Exception):
    """Base for anything the core refuses to do."""


class NotFound(CoreError):
    """The thing asked for does not exist, or is not visible to this principal.

    One error for both cases on purpose: telling an unauthorised caller that a
    resource exists but is not theirs leaks the roster of every other tenant.
    """


class Invalid(CoreError):
    """The request is malformed or violates a rule."""


class Conflict(CoreError):
    """The request collides with something already stored."""


class Forbidden(CoreError):
    """The principal is known but not permitted to do this."""


class Scope:
    """Coarse capability names a principal can hold. These gate *kinds* of
    action; row visibility is tenancy's job (scope_to_org)."""

    READ = "read"
    WRITE = "write"
    # Separate from WRITE so an ordinary write credential cannot perform an
    # administrative act - inviting a member, changing a role, deleting a tenant.
    ADMIN = "admin"


ALL_SCOPES = frozenset({Scope.READ, Scope.WRITE, Scope.ADMIN})


# The surfaces allowed to hold no organisation, and so to read across every
# tenant. One, and named out loud: scripts, seeds and migrations, which answer to
# whoever ran them. Add to this set only with the same deliberation - a staff
# console belongs here, an ordinary feature never does.
SPANNING_SURFACES = frozenset({"system"})


@dataclass(frozen=True)
class Principal:
    """The caller, resolved from whatever the façade used to authenticate."""

    subject: str
    # "ui" | "api" | "system" - how this caller arrived. Add surfaces as you add
    # façades; the name is what appears in audit lines and in Forbidden messages.
    surface: str
    scopes: frozenset = field(default_factory=frozenset)

    # Which customer's data this caller sees. None is permitted only for the
    # surfaces in SPANNING_SURFACES, which legitimately cross organisations.
    org_id: int | None = None

    # Which person/account, when there is one.
    user_id: int | None = None

    def __post_init__(self):
        """A customer-facing principal without an organisation is the bug that
        leaks one tenant's data to another, so it is rejected at construction
        rather than relied on to be filtered downstream."""
        if self.surface not in SPANNING_SURFACES and self.org_id is None:
            raise Invalid(
                f"A {self.surface} principal must be scoped to an organisation. "
                f"Only {' and '.join(sorted(SPANNING_SURFACES))} callers may span "
                "organisations."
            )

    @property
    def spans_orgs(self) -> bool:
        return self.org_id is None

    def may(self, scope: str) -> bool:
        return scope in self.scopes

    def require(self, scope: str) -> None:
        if not self.may(scope):
            raise Forbidden(
                f"This action needs the {scope!r} permission, which a {self.surface} "
                f"caller does not hold."
            )

    @classmethod
    def for_ui(cls, org_id: int, subject: str = "dashboard",
               user_id: int | None = None) -> "Principal":
        """A fully-scoped dashboard caller. For tests and bootstrap; real
        sign-ins go through auth.principal_for_user, which derives scopes from
        the membership role."""
        return cls(subject=subject, surface="ui", scopes=ALL_SCOPES,
                   org_id=org_id, user_id=user_id)

    @classmethod
    def system(cls, subject: str = "cli", org_id: int | None = None) -> "Principal":
        """Scripts, seeds and migrations - the only principal allowed to span
        organisations, and named rather than implied so that crossing tenants has
        to be said out loud."""
        return cls(subject=subject, surface="system", scopes=ALL_SCOPES,
                   org_id=org_id)


def scope_to_org(query, model, principal: Principal):
    """Narrow a query to the principal's organisation - the single place the
    tenancy rule is written down. A system caller spans organisations and the
    query comes back untouched."""
    if principal.spans_orgs:
        return query
    return query.filter(model.org_id == principal.org_id)