Skip to content

Class tigrbl_auth.orm.client.Client

tigrbl_auth.orm.client.Client

Bases: ClientBase

tenant class-attribute instance-attribute

tenant = relationship('Tenant', back_populates='clients')

grant_types class-attribute instance-attribute

grant_types = acol(
    spec=ColumnSpec(
        storage=S(
            String,
            nullable=False,
            default="authorization_code",
        ),
        field=F(),
        io=IO(),
    )
)

response_types class-attribute instance-attribute

response_types = acol(
    spec=ColumnSpec(
        storage=S(String, nullable=False, default="code"),
        field=F(),
        io=IO(),
    )
)

is_active class-attribute instance-attribute

is_active = acol(
    spec=ColumnSpec(
        storage=S(
            type_=Boolean, default=True, nullable=False
        ),
        field=F(py_type=bool),
        io=CRUD_IO,
    )
)

created_at class-attribute instance-attribute

created_at = acol(
    spec=ColumnSpec(
        storage=S(
            type_=TZDateTime,
            default=tzutcnow,
            nullable=False,
        ),
        field=F(py_type=datetime),
        io=RO_IO,
    )
)

updated_at class-attribute instance-attribute

updated_at = acol(
    spec=ColumnSpec(
        storage=S(
            type_=TZDateTime,
            default=tzutcnow,
            onupdate=tzutcnow,
            nullable=False,
        ),
        field=F(py_type=datetime),
        io=RO_IO,
    )
)

id class-attribute instance-attribute

id = acol(
    spec=ColumnSpec(
        storage=S(
            type_=PgUUID(as_uuid=True),
            primary_key=True,
            default=uuid4,
        ),
        field=F(
            py_type=UUID,
            constraints={"examples": [uuid_example]},
        ),
        io=RO_IO,
    )
)

metadata class-attribute instance-attribute

metadata = MetaData(
    naming_convention={
        "pk": "pk_%(table_name)s",
        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
        "ix": "ix_%(table_name)s_%(column_0_name)s",
        "uq": "uq_%(table_name)s_%(column_0_name)s",
        "ck": "ck_%(table_name)s_%(column_0_name)s_%(constraint_type)s",
    }
)

client_secret_hash class-attribute instance-attribute

client_secret_hash = acol(
    storage=S(LargeBinary(60), nullable=False),
    field=F(),
    io=IO(in_verbs=("create",)),
)

redirect_uris class-attribute instance-attribute

redirect_uris = acol(
    storage=S(String, nullable=False),
    field=F(constraints={"max_length": 1000}),
    io=IO(),
)

new classmethod

new(tenant_id, client_id, client_secret, redirects)
Source code in tigrbl_auth/orm/client.py
 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
@classmethod
def new(
    cls,
    tenant_id: uuid.UUID,
    client_id: str,
    client_secret: str,
    redirects: list[str],
):
    if not _CLIENT_ID_RE.fullmatch(client_id):
        raise ValueError("invalid client_id format")
    if settings.enforce_rfc8252:
        for uri in redirects:
            parsed = urlparse(uri)
            if is_native_redirect_uri(uri):
                validate_native_redirect_uri(uri)
            elif parsed.scheme == "http":
                raise ValueError(
                    f"redirect URI not permitted for native apps per RFC 8252: {RFC8252_SPEC_URL}"
                )
    secret_hash = hash_pw(client_secret)
    try:
        obj_id: uuid.UUID | str = uuid.UUID(client_id)
    except ValueError:
        obj_id = client_id
    return cls(
        tenant_id=tenant_id,
        id=obj_id,
        client_secret_hash=secret_hash,
        redirect_uris=" ".join(redirects),
    )

register async

register(ctx)
Source code in tigrbl_auth/orm/client.py
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@op_ctx(
    alias="register",
    target="create",
    arity="collection",
    status_code=status.HTTP_201_CREATED,
)
async def register(cls, ctx):
    import secrets

    from urllib.parse import urlparse

    payload = ctx.get("payload") or {}
    redirects = payload.get("redirect_uris") or []
    if isinstance(redirects, list):
        for uri in redirects:
            parsed = urlparse(uri)
            if parsed.scheme != "https" and parsed.hostname not in {
                "localhost",
                "127.0.0.1",
                "::1",
            }:
                raise HTTPException(
                    status.HTTP_400_BAD_REQUEST,
                    "redirect URIs must use https scheme",
                )
        payload["redirect_uris"] = " ".join(redirects)
    grant_types = payload.get("grant_types")
    if isinstance(grant_types, list):
        payload["grant_types"] = " ".join(grant_types)
    else:
        payload.setdefault("grant_types", "authorization_code")
    response_types = payload.get("response_types")
    if isinstance(response_types, list):
        payload["response_types"] = " ".join(response_types)
    else:
        payload.setdefault("response_types", "code")
    client_id = payload.get("client_id") or secrets.token_urlsafe(16)
    payload["id"] = client_id
    secret = payload.get("client_secret") or secrets.token_urlsafe(32)
    payload["client_secret"] = secret
    obj = await cls.handlers.create.core({"payload": payload})
    return {
        "client_id": str(obj.id),
        "client_secret": secret,
        "redirect_uris": obj.redirect_uris.split(),
        "grant_types": obj.grant_types.split(),
        "response_types": obj.response_types.split(),
    }

verify_secret

verify_secret(plain)
Source code in tigrbl_auth/orm/client.py
159
160
161
162
def verify_secret(self, plain: str) -> bool:
    from ..crypto import verify_pw  # local import to avoid cycle

    return verify_pw(plain, self.client_secret_hash)

is_visible staticmethod

is_visible(obj, ctx)
Source code in tigrbl/orm/mixins/tenant_bound.py
110
111
112
@staticmethod
def is_visible(obj, ctx) -> bool:
    return getattr(obj, "tenant_id", None) == _ctx_tenant_id(ctx)

tenant_id

tenant_id()
Source code in tigrbl/orm/mixins/tenant_bound.py
 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
@declared_attr
def tenant_id(cls) -> Mapped[UUID]:
    pol = getattr(cls, TIGRBL_TENANT_POLICY_ATTR, TenantPolicy.CLIENT_SET)
    schema = _infer_schema(cls, default="public")

    in_verbs = (
        ("create", "update", "replace")
        if pol == TenantPolicy.CLIENT_SET
        else ("create",)
    )
    io = IO(
        in_verbs=in_verbs,
        out_verbs=("read", "list"),
        mutable_verbs=in_verbs,
    )

    spec = ColumnSpec(
        storage=S(
            type_=PgUUID(as_uuid=True),
            fk=ForeignKeySpec(target=f"{schema}.tenants.id"),
            nullable=False,
            index=True,
        ),
        field=F(py_type=UUID),
        io=io,
    )
    return acol(spec=spec)