Skip to content

Users and accounts

octomate.managers.user

The cross-channel user registry: users, the channel profiles they own, and profile linking.

PROFILE_FIELDS module-attribute 

PROFILE_FIELDS = {
    "name",
    "nickname",
    "gender",
    "age",
    "title",
}

LinkProfileUnavailable

Bases: RuntimeError

Profile linking cannot start: local auth or the callback origin is not configured.

Source code in octomate/managers/user.py 
26
27
28
class LinkProfileUnavailable(RuntimeError):
    """Profile linking cannot start: local auth or the callback origin is not
    configured."""

InvalidLinkProfile

Bases: ValueError

A profile-link ticket that is invalid, expired or already used.

Source code in octomate/managers/user.py 
31
32
33
34
35
class InvalidLinkProfile(ValueError):
    """A profile-link ticket that is invalid, expired or already used."""

    def __init__(self) -> None:
        super().__init__("This profile link is invalid, expired, or already used")

ProfileAlreadyLinked

Bases: ValueError

The channel profile is already linked to an account.

Source code in octomate/managers/user.py 
38
39
40
41
42
class ProfileAlreadyLinked(ValueError):
    """The channel profile is already linked to an account."""

    def __init__(self) -> None:
        super().__init__("This channel profile is already linked")

ProfileNotLinked

Bases: ValueError

The profile is not linked to the requesting account.

Source code in octomate/managers/user.py 
45
46
47
48
49
class ProfileNotLinked(ValueError):
    """The profile is not linked to the requesting account."""

    def __init__(self) -> None:
        super().__init__("This profile is not linked to your account")

UserManager

Bases: Manager, Locks[tuple[str, str] | UUID7]

The persisted cross-channel user registry and verified profile links.

Source code in octomate/managers/user.py 
 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
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
class UserManager(Manager, Locks[tuple[str, str] | UUID7]):
    """The persisted cross-channel user registry and verified profile links."""

    def __init__(
        self,
        *,
        authorization_base_uri: AnyHttpUrl | None = None,
        authorization_lifetime: timedelta = timedelta(minutes=10),
    ) -> None:
        self.authorization_base_uri = authorization_base_uri
        self.authorization_lifetime = authorization_lifetime

    async def unlink_profile(self, user: User, profile_id: UUID7) -> None:
        async with self.lock(profile_id), async_session() as session:
            profile = await session.get(UserProfile, profile_id)
            if profile is None or profile.user_id != user.id:
                raise ProfileNotLinked
            profile.user_id = None
            pending = await session.one_or_none(
                LinkProfileSession,
                expressions=[LinkProfileSession["profile_id"] == profile_id],
            )
            if pending is not None:
                pending.consumed_at = datetime.now(UTC)
            await session.commit()

    async def owner(self, profile: UserProfile) -> User | None:
        """Return the registered owner of ``profile``, or ``None`` for a visitor."""
        user_id = profile.user_id
        if user_id is None:
            return None
        async with async_session() as session:
            user = await session.get(User, user_id)
        if user is None:
            raise ValueError(f"unknown user {user_id}")
        return user

    async def linked_profiles(
        self,
        profile: UserProfile,
    ) -> list[UserProfile]:
        """This human's other channel identities — the same person on another
        platform, reachable precisely because the registry links the accounts.

        Empty for a visitor: only a registered `User` links identities, so an
        unregistered account is one account and can be followed nowhere. `owner`
        answers "whose is this"; this answers "where else are they".
        """
        user = await self.owner(profile)
        if user is None:
            return []
        async with async_session() as session:
            return list(
                await session.list(
                    UserProfile,
                    limit=None,
                    expressions=[
                        UserProfile["user_id"] == user.id,
                        UserProfile["channel_tentacle_id"]
                        != profile.channel_tentacle_id,
                    ],
                )
            )

    async def native_profile(self, runtime: str, username: str) -> UserProfile | None:
        """A transient anchor for `username`'s native session on `runtime`'s
        pseudo-channel, or None for an unknown username.

        Never persisted: a native session's identity comes from its verified
        bearer, not from a claimed row, so the profile exists only to give the
        linked-profile walk its starting point — owned like a stored profile,
        and standing on a channel id no channel ever resolves.
        """
        async with async_session() as session:
            user = await session.one_or_none(
                User, expressions=[User["username"] == username]
            )
        if user is None:
            return None
        return UserProfile(
            channel_tentacle_id=runtime,
            channel_user_id=username,
            user_id=user.id,
            user=Relation(user),
            name=user.name,
            nickname=user.nickname,
        )

    async def profile(
        self, channel_tentacle_id: str, channel_user_id: str
    ) -> UserProfile | None:
        """The stored profile for a channel identity, or ``None`` if never observed.

        The read-only sibling of ``ensure_profile``, for callers that hold only a
        channel identity — e.g. a deferred batch resuming long after the message
        that carried the sender's profile snapshot."""
        async with async_session() as session:
            return await session.one_or_none(
                UserProfile,
                expressions=[
                    UserProfile["channel_tentacle_id"] == channel_tentacle_id,
                    UserProfile["channel_user_id"] == channel_user_id,
                ],
            )

    async def ensure_profile(
        self, channel_tentacle_id: str, observed: UserProfile
    ) -> UserProfile:
        """Persist the latest channel snapshot and return its registry profile.

        First sight creates an ownerless visitor profile. An existing profile
        keeps its ownership while observations refresh its display fields. An
        `observed.user_id` is the one exception: no channel ever sets it —
        only a verified bearer's transient anchor (`native_profile`) carries
        one — so an identity that arrives owned stays owned.
        """
        key = (channel_tentacle_id, observed.channel_user_id)
        async with self.lock(key), async_session() as session:
            profile = await session.one_or_none(
                UserProfile,
                expressions=[
                    UserProfile["channel_tentacle_id"] == channel_tentacle_id,
                    UserProfile["channel_user_id"] == observed.channel_user_id,
                ],
            )
            if profile is None:
                profile = UserProfile(
                    channel_tentacle_id=channel_tentacle_id,
                    channel_user_id=observed.channel_user_id,
                    user_id=observed.user_id,
                    **observed.model_dump(include=PROFILE_FIELDS),
                )
                session.add(profile)
            else:
                for name in PROFILE_FIELDS:
                    setattr(profile, name, getattr(observed, name))
                if observed.user_id is not None:
                    profile.user_id = observed.user_id

            await session.commit()

        return profile

    @staticmethod
    def hash_link_token(token: SecretStr) -> SecretStr:
        return SecretStr(hashlib.sha256(token.get_secret_value().encode()).hexdigest())

    async def link_verified_profile(self, profile: UserProfile, user: User) -> None:
        """Link an OAuth-verified profile without replacing an existing owner."""
        async with self.lock(profile.id), async_session() as session:
            stored = await session.get(UserProfile, profile.id)
            if stored is None or await session.get(User, user.id) is None:
                raise InvalidLinkProfile
            if stored.user_id is not None and stored.user_id != user.id:
                raise ProfileAlreadyLinked
            stored.user_id = user.id
            pending = await session.one_or_none(
                LinkProfileSession,
                expressions=[LinkProfileSession["profile_id"] == stored.id],
            )
            if pending is not None:
                pending.consumed_at = datetime.now(UTC)
            await session.commit()

    async def start_link_profile(
        self, profile: UserProfile
    ) -> LinkProfileAuthorization:
        """Replace this ownerless profile's pending ticket and return its private URL."""
        if self.authorization_base_uri is None:
            raise LinkProfileUnavailable(
                "Configure local auth and oauth.callback_base_uri before linking channel profiles"
            )
        token = SecretStr(secrets.token_urlsafe(32))
        now = datetime.now(UTC)
        async with self.lock(profile.id), async_session() as session:
            stored = await session.get(UserProfile, profile.id)
            if stored is None:
                raise InvalidLinkProfile
            if stored.user_id is not None:
                raise ProfileAlreadyLinked
            pending = await session.one_or_none(
                LinkProfileSession,
                expressions=[LinkProfileSession["profile_id"] == stored.id],
            )
            if pending is None:
                pending = LinkProfileSession(
                    profile_id=stored.id,
                    token_hash=self.hash_link_token(token),
                    expires_at=now + self.authorization_lifetime,
                )
                session.add(pending)
            else:
                pending.token_hash = self.hash_link_token(token)
                pending.expires_at = now + self.authorization_lifetime
                pending.created_at = now
                pending.consumed_at = None
            await session.commit()

        root = str(self.authorization_base_uri).rstrip("/")
        return LinkProfileAuthorization(
            profile=stored,
            authorization_uri=AnyHttpUrl(
                f"{root}/#link-profile={token.get_secret_value()}"
            ),
            expires_at=now + self.authorization_lifetime,
        )

    async def inspect_link_profile(self, token: SecretStr) -> LinkProfileInfo:
        token_hash = self.hash_link_token(token)
        async with async_session() as session:
            pending = await session.one_or_none(
                LinkProfileSession,
                expressions=[
                    LinkProfileSession["token_hash"] == token_hash,
                    LinkProfileSession["consumed_at"].is_(None),
                    LinkProfileSession["expires_at"] > datetime.now(UTC),
                ],
            )
            if pending is None:
                raise InvalidLinkProfile
            profile = await session.get(UserProfile, pending.profile_id)
            if profile is None:
                raise InvalidLinkProfile
            if profile.user_id is not None:
                raise ProfileAlreadyLinked
            return LinkProfileInfo(profile=profile, expires_at=pending.expires_at)

    async def confirm_link_profile(self, token: SecretStr, user: User) -> UserProfile:
        """Consume a valid ticket and link its exact profile to ``user`` once."""
        token_hash = self.hash_link_token(token)
        async with async_session() as session:
            found = await session.one_or_none(
                LinkProfileSession,
                expressions=[LinkProfileSession["token_hash"] == token_hash],
            )
        if found is None:
            raise InvalidLinkProfile

        async with self.lock(found.profile_id), async_session() as session:
            pending = await session.one_or_none(
                LinkProfileSession,
                expressions=[
                    LinkProfileSession["token_hash"] == token_hash,
                    LinkProfileSession["consumed_at"].is_(None),
                    LinkProfileSession["expires_at"] > datetime.now(UTC),
                ],
            )
            if pending is None:
                raise InvalidLinkProfile
            profile = await session.get(UserProfile, pending.profile_id)
            if profile is None or await session.get(User, user.id) is None:
                raise InvalidLinkProfile
            if profile.user_id is not None:
                raise ProfileAlreadyLinked
            profile.user_id = user.id
            pending.consumed_at = datetime.now(UTC)
            try:
                await session.flush()
            except StaleDataError as error:
                raise InvalidLinkProfile from error
            await session.commit()
            return profile

authorization_base_uri instance-attribute 

authorization_base_uri = authorization_base_uri

authorization_lifetime instance-attribute 

authorization_lifetime = authorization_lifetime

unlink_profile(user: User, profile_id: UUID7) -> None
Source code in octomate/managers/user.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
async def unlink_profile(self, user: User, profile_id: UUID7) -> None:
    async with self.lock(profile_id), async_session() as session:
        profile = await session.get(UserProfile, profile_id)
        if profile is None or profile.user_id != user.id:
            raise ProfileNotLinked
        profile.user_id = None
        pending = await session.one_or_none(
            LinkProfileSession,
            expressions=[LinkProfileSession["profile_id"] == profile_id],
        )
        if pending is not None:
            pending.consumed_at = datetime.now(UTC)
        await session.commit()

owner async 

owner(profile: UserProfile) -> User | None

Return the registered owner of profile, or None for a visitor.

Source code in octomate/managers/user.py 
78
79
80
81
82
83
84
85
86
87
async def owner(self, profile: UserProfile) -> User | None:
    """Return the registered owner of ``profile``, or ``None`` for a visitor."""
    user_id = profile.user_id
    if user_id is None:
        return None
    async with async_session() as session:
        user = await session.get(User, user_id)
    if user is None:
        raise ValueError(f"unknown user {user_id}")
    return user

linked_profiles async 

linked_profiles(profile: UserProfile) -> list[UserProfile]

This human's other channel identities — the same person on another platform, reachable precisely because the registry links the accounts.

Empty for a visitor: only a registered User links identities, so an unregistered account is one account and can be followed nowhere. owner answers "whose is this"; this answers "where else are they".

Source code in octomate/managers/user.py 
 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
async def linked_profiles(
    self,
    profile: UserProfile,
) -> list[UserProfile]:
    """This human's other channel identities — the same person on another
    platform, reachable precisely because the registry links the accounts.

    Empty for a visitor: only a registered `User` links identities, so an
    unregistered account is one account and can be followed nowhere. `owner`
    answers "whose is this"; this answers "where else are they".
    """
    user = await self.owner(profile)
    if user is None:
        return []
    async with async_session() as session:
        return list(
            await session.list(
                UserProfile,
                limit=None,
                expressions=[
                    UserProfile["user_id"] == user.id,
                    UserProfile["channel_tentacle_id"]
                    != profile.channel_tentacle_id,
                ],
            )
        )

native_profile async 

native_profile(
    runtime: str, username: str
) -> UserProfile | None

A transient anchor for username's native session on runtime's pseudo-channel, or None for an unknown username.

Never persisted: a native session's identity comes from its verified bearer, not from a claimed row, so the profile exists only to give the linked-profile walk its starting point — owned like a stored profile, and standing on a channel id no channel ever resolves.

Source code in octomate/managers/user.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def native_profile(self, runtime: str, username: str) -> UserProfile | None:
    """A transient anchor for `username`'s native session on `runtime`'s
    pseudo-channel, or None for an unknown username.

    Never persisted: a native session's identity comes from its verified
    bearer, not from a claimed row, so the profile exists only to give the
    linked-profile walk its starting point — owned like a stored profile,
    and standing on a channel id no channel ever resolves.
    """
    async with async_session() as session:
        user = await session.one_or_none(
            User, expressions=[User["username"] == username]
        )
    if user is None:
        return None
    return UserProfile(
        channel_tentacle_id=runtime,
        channel_user_id=username,
        user_id=user.id,
        user=Relation(user),
        name=user.name,
        nickname=user.nickname,
    )

profile async 

profile(
    channel_tentacle_id: str, channel_user_id: str
) -> UserProfile | None

The stored profile for a channel identity, or None if never observed.

The read-only sibling of ensure_profile, for callers that hold only a channel identity — e.g. a deferred batch resuming long after the message that carried the sender's profile snapshot.

Source code in octomate/managers/user.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
async def profile(
    self, channel_tentacle_id: str, channel_user_id: str
) -> UserProfile | None:
    """The stored profile for a channel identity, or ``None`` if never observed.

    The read-only sibling of ``ensure_profile``, for callers that hold only a
    channel identity — e.g. a deferred batch resuming long after the message
    that carried the sender's profile snapshot."""
    async with async_session() as session:
        return await session.one_or_none(
            UserProfile,
            expressions=[
                UserProfile["channel_tentacle_id"] == channel_tentacle_id,
                UserProfile["channel_user_id"] == channel_user_id,
            ],
        )

ensure_profile async 

ensure_profile(
    channel_tentacle_id: str, observed: UserProfile
) -> UserProfile

Persist the latest channel snapshot and return its registry profile.

First sight creates an ownerless visitor profile. An existing profile keeps its ownership while observations refresh its display fields. An observed.user_id is the one exception: no channel ever sets it — only a verified bearer's transient anchor (native_profile) carries one — so an identity that arrives owned stays owned.

Source code in octomate/managers/user.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def ensure_profile(
    self, channel_tentacle_id: str, observed: UserProfile
) -> UserProfile:
    """Persist the latest channel snapshot and return its registry profile.

    First sight creates an ownerless visitor profile. An existing profile
    keeps its ownership while observations refresh its display fields. An
    `observed.user_id` is the one exception: no channel ever sets it —
    only a verified bearer's transient anchor (`native_profile`) carries
    one — so an identity that arrives owned stays owned.
    """
    key = (channel_tentacle_id, observed.channel_user_id)
    async with self.lock(key), async_session() as session:
        profile = await session.one_or_none(
            UserProfile,
            expressions=[
                UserProfile["channel_tentacle_id"] == channel_tentacle_id,
                UserProfile["channel_user_id"] == observed.channel_user_id,
            ],
        )
        if profile is None:
            profile = UserProfile(
                channel_tentacle_id=channel_tentacle_id,
                channel_user_id=observed.channel_user_id,
                user_id=observed.user_id,
                **observed.model_dump(include=PROFILE_FIELDS),
            )
            session.add(profile)
        else:
            for name in PROFILE_FIELDS:
                setattr(profile, name, getattr(observed, name))
            if observed.user_id is not None:
                profile.user_id = observed.user_id

        await session.commit()

    return profile

hash_link_token(token: SecretStr) -> SecretStr
Source code in octomate/managers/user.py 
195
196
197
@staticmethod
def hash_link_token(token: SecretStr) -> SecretStr:
    return SecretStr(hashlib.sha256(token.get_secret_value().encode()).hexdigest())

link_verified_profile(
    profile: UserProfile, user: User
) -> None

Link an OAuth-verified profile without replacing an existing owner.

Source code in octomate/managers/user.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
async def link_verified_profile(self, profile: UserProfile, user: User) -> None:
    """Link an OAuth-verified profile without replacing an existing owner."""
    async with self.lock(profile.id), async_session() as session:
        stored = await session.get(UserProfile, profile.id)
        if stored is None or await session.get(User, user.id) is None:
            raise InvalidLinkProfile
        if stored.user_id is not None and stored.user_id != user.id:
            raise ProfileAlreadyLinked
        stored.user_id = user.id
        pending = await session.one_or_none(
            LinkProfileSession,
            expressions=[LinkProfileSession["profile_id"] == stored.id],
        )
        if pending is not None:
            pending.consumed_at = datetime.now(UTC)
        await session.commit()

start_link_profile(
    profile: UserProfile,
) -> LinkProfileAuthorization

Replace this ownerless profile's pending ticket and return its private URL.

Source code in octomate/managers/user.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
async def start_link_profile(
    self, profile: UserProfile
) -> LinkProfileAuthorization:
    """Replace this ownerless profile's pending ticket and return its private URL."""
    if self.authorization_base_uri is None:
        raise LinkProfileUnavailable(
            "Configure local auth and oauth.callback_base_uri before linking channel profiles"
        )
    token = SecretStr(secrets.token_urlsafe(32))
    now = datetime.now(UTC)
    async with self.lock(profile.id), async_session() as session:
        stored = await session.get(UserProfile, profile.id)
        if stored is None:
            raise InvalidLinkProfile
        if stored.user_id is not None:
            raise ProfileAlreadyLinked
        pending = await session.one_or_none(
            LinkProfileSession,
            expressions=[LinkProfileSession["profile_id"] == stored.id],
        )
        if pending is None:
            pending = LinkProfileSession(
                profile_id=stored.id,
                token_hash=self.hash_link_token(token),
                expires_at=now + self.authorization_lifetime,
            )
            session.add(pending)
        else:
            pending.token_hash = self.hash_link_token(token)
            pending.expires_at = now + self.authorization_lifetime
            pending.created_at = now
            pending.consumed_at = None
        await session.commit()

    root = str(self.authorization_base_uri).rstrip("/")
    return LinkProfileAuthorization(
        profile=stored,
        authorization_uri=AnyHttpUrl(
            f"{root}/#link-profile={token.get_secret_value()}"
        ),
        expires_at=now + self.authorization_lifetime,
    )

inspect_link_profile(token: SecretStr) -> LinkProfileInfo
Source code in octomate/managers/user.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
async def inspect_link_profile(self, token: SecretStr) -> LinkProfileInfo:
    token_hash = self.hash_link_token(token)
    async with async_session() as session:
        pending = await session.one_or_none(
            LinkProfileSession,
            expressions=[
                LinkProfileSession["token_hash"] == token_hash,
                LinkProfileSession["consumed_at"].is_(None),
                LinkProfileSession["expires_at"] > datetime.now(UTC),
            ],
        )
        if pending is None:
            raise InvalidLinkProfile
        profile = await session.get(UserProfile, pending.profile_id)
        if profile is None:
            raise InvalidLinkProfile
        if profile.user_id is not None:
            raise ProfileAlreadyLinked
        return LinkProfileInfo(profile=profile, expires_at=pending.expires_at)

confirm_link_profile(
    token: SecretStr, user: User
) -> UserProfile

Consume a valid ticket and link its exact profile to user once.

Source code in octomate/managers/user.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
async def confirm_link_profile(self, token: SecretStr, user: User) -> UserProfile:
    """Consume a valid ticket and link its exact profile to ``user`` once."""
    token_hash = self.hash_link_token(token)
    async with async_session() as session:
        found = await session.one_or_none(
            LinkProfileSession,
            expressions=[LinkProfileSession["token_hash"] == token_hash],
        )
    if found is None:
        raise InvalidLinkProfile

    async with self.lock(found.profile_id), async_session() as session:
        pending = await session.one_or_none(
            LinkProfileSession,
            expressions=[
                LinkProfileSession["token_hash"] == token_hash,
                LinkProfileSession["consumed_at"].is_(None),
                LinkProfileSession["expires_at"] > datetime.now(UTC),
            ],
        )
        if pending is None:
            raise InvalidLinkProfile
        profile = await session.get(UserProfile, pending.profile_id)
        if profile is None or await session.get(User, user.id) is None:
            raise InvalidLinkProfile
        if profile.user_id is not None:
            raise ProfileAlreadyLinked
        profile.user_id = user.id
        pending.consumed_at = datetime.now(UTC)
        try:
            await session.flush()
        except StaleDataError as error:
            raise InvalidLinkProfile from error
        await session.commit()
        return profile

octomate.managers.auth

Invite-only registration, local passwords, sessions and personal API keys.

InvalidCredentials

Bases: ValueError

A sign-in, refresh or password change whose credentials are wrong or expired.

Source code in octomate/managers/auth.py 
31
32
33
34
35
36
class InvalidCredentials(ValueError):
    """A sign-in, refresh or password change whose credentials are wrong or
    expired."""

    def __init__(self) -> None:
        super().__init__("Invalid or expired credentials")

UsernameUnavailable

Bases: ValueError

A registration whose username is already taken.

Source code in octomate/managers/auth.py 
39
40
41
42
43
class UsernameUnavailable(ValueError):
    """A registration whose username is already taken."""

    def __init__(self) -> None:
        super().__init__("This username is already taken")

AuthManager

Bases: Manager, Locks[str]

Invite-only registration, local passwords, and user credentials.

Source code in octomate/managers/auth.py 
 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
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
class AuthManager(Manager, Locks[str]):
    """Invite-only registration, local passwords, and user credentials."""

    def __init__(self, config: AuthConfig) -> None:
        self.config: AuthConfig = config
        self.password_hasher: PasswordHasher = PasswordHasher()
        self.password_limiter: CapacityLimiter = CapacityLimiter(2)
        # Unknown and unenrolled users still pay the password verification cost.
        self.dummy_password_hash: str = self.password_hasher.hash(
            secrets.token_urlsafe(32)
        )

    async def hash_password(self, password: SecretStr) -> SecretStr:
        return SecretStr(
            await to_thread.run_sync(
                self.password_hasher.hash,
                password.get_secret_value(),
                limiter=self.password_limiter,
            )
        )

    @staticmethod
    def hash_token(token: SecretStr, salt: SecretStr) -> SecretStr:
        return SecretStr(
            hashlib.sha256(
                (token.get_secret_value() + salt.get_secret_value()).encode()
            ).hexdigest()
        )

    async def invite(self) -> SecretStr:
        """Issue an anonymous code that admits one new account."""
        token = SecretStr(secrets.token_urlsafe(32))
        async with async_session() as session:
            session.add(
                UserInvitation(
                    token_hash=SecretStr(
                        hashlib.sha256(token.get_secret_value().encode()).hexdigest()
                    ),
                    expires_at=datetime.now(UTC) + self.config.invitation_lifetime,
                )
            )
            await session.commit()
            return token

    async def register(
        self, username: str, password: SecretStr, invitation: SecretStr, *, name: str
    ) -> User:
        if not username or username != username.strip() or len(username) > 100:
            raise ValueError(
                "Username must contain 1-100 characters without surrounding spaces"
            )
        password = NewPasswordAdapter.validate_python(password)
        if not 1 <= len(name) <= 100:
            raise ValueError("Display name must contain 1-100 characters")
        async with self.lock(username), async_session() as session:
            token_hash = SecretStr(
                hashlib.sha256(invitation.get_secret_value().encode()).hexdigest()
            )
            current = await session.one_or_none(
                UserInvitation,
                expressions=[
                    UserInvitation["token_hash"] == token_hash,
                    UserInvitation["consumed_at"].is_(None),
                    UserInvitation["expires_at"] > datetime.now(UTC),
                ],
            )
            if current is None:
                raise InvalidCredentials
            if (
                await session.one_or_none(
                    User, expressions=[User["username"] == username]
                )
                is not None
            ):
                raise UsernameUnavailable
            user = User(
                username=username,
                name=name,
                password_hash=await self.hash_password(password),
            )
            session.add(user)
            now = datetime.now(UTC)
            if current.expires_at <= now:
                raise InvalidCredentials
            current.consumed_at = now
            try:
                await session.flush()
            except StaleDataError as error:
                raise InvalidCredentials from error
            await session.commit()
            return user

    async def login(self, username: str, password: SecretStr) -> SessionTokens:
        async with async_session() as session:
            user = await session.one_or_none(
                User, expressions=[User["username"] == username]
            )
            password_hash = (
                user.password_hash.get_secret_value()
                if user is not None and user.password_hash is not None
                else self.dummy_password_hash
            )
            try:
                await to_thread.run_sync(
                    self.password_hasher.verify,
                    password_hash,
                    password.get_secret_value(),
                    limiter=self.password_limiter,
                )
            except VerifyMismatchError as error:
                raise InvalidCredentials from error
            if user is None or user.password_hash is None:
                raise InvalidCredentials
            if self.password_hasher.check_needs_rehash(password_hash):
                user.password_hash = await self.hash_password(password)

            now = datetime.now(UTC)
            tokens = SessionTokens(
                session_id=uuid7(),
                user_id=user.id,
                access_token=SecretStr(secrets.token_urlsafe(32)),
                refresh_token=SecretStr(secrets.token_urlsafe(32)),
                access_expires_at=now + self.config.access_token_lifetime,
                refresh_expires_at=now + self.config.session_lifetime,
            )
            session.add(
                UserSession(
                    id=tokens.session_id,
                    user_id=user.id,
                    access_token_hash=self.hash_token(
                        tokens.access_token, self.config.access_token_salt
                    ),
                    refresh_token_hash=self.hash_token(
                        tokens.refresh_token, self.config.refresh_token_salt
                    ),
                    access_expires_at=tokens.access_expires_at,
                    refresh_expires_at=tokens.refresh_expires_at,
                )
            )
            await session.commit()
            return tokens

    async def authenticate_session(self, token: SecretStr) -> UserSession | None:
        now = datetime.now(UTC)
        async with async_session() as session:
            return await session.one_or_none(
                UserSession,
                expressions=[
                    UserSession["access_token_hash"]
                    == self.hash_token(token, self.config.access_token_salt),
                    UserSession["revoked_at"].is_(None),
                    UserSession["access_expires_at"] > now,
                    UserSession["refresh_expires_at"] > now,
                ],
            )

    async def set_password(
        self,
        username: str,
        password: SecretStr,
        *,
        current_password: SecretStr | None = None,
    ) -> None:
        """Replace a password and end browser sessions; HTTP callers must supply the old password."""
        password = NewPasswordAdapter.validate_python(password)
        async with self.lock(username), async_session() as session:
            user = await session.one_or_none(
                User, expressions=[User["username"] == username]
            )
            if user is None or user.password_hash is None:
                raise InvalidCredentials
            if current_password is not None:
                try:
                    await to_thread.run_sync(
                        self.password_hasher.verify,
                        user.password_hash.get_secret_value(),
                        current_password.get_secret_value(),
                        limiter=self.password_limiter,
                    )
                except VerifyMismatchError as error:
                    raise InvalidCredentials from error
            user.password_hash = await self.hash_password(password)
            now = datetime.now(UTC)
            for current in await session.list(
                UserSession,
                expressions=[
                    UserSession["user_id"] == user.id,
                    UserSession["revoked_at"].is_(None),
                ],
                limit=None,
            ):
                current.revoked_at = now
            await session.commit()

    async def logout(
        self, access_token: SecretStr | None, refresh_token: SecretStr | None
    ) -> None:
        expressions = []
        if access_token is not None:
            expressions.append(
                UserSession["access_token_hash"]
                == self.hash_token(access_token, self.config.access_token_salt)
            )
        if refresh_token is not None:
            expressions.append(
                UserSession["refresh_token_hash"]
                == self.hash_token(refresh_token, self.config.refresh_token_salt)
            )
        if not expressions:
            return
        async with async_session() as session:
            for current in await session.list(
                UserSession,
                expressions=[or_(*expressions), UserSession["revoked_at"].is_(None)],
                limit=None,
            ):
                current.revoked_at = datetime.now(UTC)
            await session.commit()

    async def refresh_session(self, token: SecretStr) -> SessionTokens:
        now = datetime.now(UTC)
        async with async_session() as session:
            current = await session.one_or_none(
                UserSession,
                expressions=[
                    UserSession["refresh_token_hash"]
                    == self.hash_token(token, self.config.refresh_token_salt),
                    UserSession["revoked_at"].is_(None),
                    UserSession["refresh_expires_at"] > now,
                ],
            )
            if current is None:
                raise InvalidCredentials
            deadline = current.refresh_expires_at
            tokens = SessionTokens(
                session_id=current.id,
                user_id=current.user_id,
                access_token=SecretStr(secrets.token_urlsafe(32)),
                refresh_token=SecretStr(secrets.token_urlsafe(32)),
                access_expires_at=min(
                    now + self.config.access_token_lifetime, deadline
                ),
                refresh_expires_at=deadline,
            )
            current.access_token_hash = self.hash_token(
                tokens.access_token, self.config.access_token_salt
            )
            current.refresh_token_hash = self.hash_token(
                tokens.refresh_token, self.config.refresh_token_salt
            )
            current.access_expires_at = tokens.access_expires_at
            try:
                await session.flush()
            except StaleDataError as error:
                raise InvalidCredentials from error
            await session.commit()
            return tokens

    async def revoke_session(self, user_id: UUID7, session_id: UUID7) -> None:
        async with async_session() as session:
            current = await session.get(UserSession, session_id)
            if current is None or current.user_id != user_id:
                raise InvalidCredentials
            if current.revoked_at is None:
                current.revoked_at = datetime.now(UTC)
                try:
                    await session.flush()
                except StaleDataError as error:
                    raise InvalidCredentials from error
                await session.commit()

    async def create_api_key(
        self,
        user_id: UUID7,
        *,
        name: str,
        scopes: list[ApiKeyScope],
        expires_at: AwareDatetime | None = None,
    ) -> IssuedApiKey:
        """Issue for a verified user; the HTTP caller must establish that identity."""
        token = SecretStr(self.config.api_key_prefix + secrets.token_urlsafe(32))
        key = UserApiKey(
            user_id=user_id,
            name=name,
            scopes=scopes,
            expires_at=expires_at,
            key_prefix=token.get_secret_value()[:12],
            key_hash=self.hash_token(token, self.config.api_key_salt),
        )
        if key.expires_at is not None and key.expires_at <= datetime.now(UTC):
            raise ValueError("API key expiry must be in the future")
        async with async_session() as session:
            if await session.get(User, user_id) is None:
                raise InvalidCredentials
            session.add(key)
            await session.commit()
        return IssuedApiKey(key=key, token=token)

    async def authenticate_api_key(
        self, token: SecretStr, *, scope: ApiKeyScope
    ) -> UserApiKey | None:
        async with async_session() as session:
            key = await session.one_or_none(
                UserApiKey,
                expressions=[
                    UserApiKey["key_hash"]
                    == self.hash_token(token, self.config.api_key_salt),
                    UserApiKey["revoked_at"].is_(None),
                    UserApiKey["expires_at"].is_(None)
                    | (UserApiKey["expires_at"] > datetime.now(UTC)),
                ],
            )
        return key if key is not None and scope in key.scopes else None

    async def list_api_keys(self, user_id: UUID7) -> list[UserApiKey]:
        async with async_session() as session:
            return list(
                await session.list(
                    UserApiKey,
                    expressions=[UserApiKey["user_id"] == user_id],
                    limit=None,
                )
            )

    async def revoke_api_key(self, user_id: UUID7, key_id: UUID7) -> None:
        async with async_session() as session:
            key = await session.get(UserApiKey, key_id)
            if key is None or key.user_id != user_id:
                raise InvalidCredentials
            if key.revoked_at is None:
                key.revoked_at = datetime.now(UTC)
                await session.commit()

config instance-attribute 

config: AuthConfig = config

password_hasher instance-attribute 

password_hasher: PasswordHasher = PasswordHasher()

password_limiter instance-attribute 

password_limiter: CapacityLimiter = CapacityLimiter(2)

dummy_password_hash instance-attribute 

dummy_password_hash: str = self.password_hasher.hash(
    secrets.token_urlsafe(32)
)

hash_password async 

hash_password(password: SecretStr) -> SecretStr
Source code in octomate/managers/auth.py 
58
59
60
61
62
63
64
65
async def hash_password(self, password: SecretStr) -> SecretStr:
    return SecretStr(
        await to_thread.run_sync(
            self.password_hasher.hash,
            password.get_secret_value(),
            limiter=self.password_limiter,
        )
    )

hash_token staticmethod 

hash_token(token: SecretStr, salt: SecretStr) -> SecretStr
Source code in octomate/managers/auth.py 
67
68
69
70
71
72
73
@staticmethod
def hash_token(token: SecretStr, salt: SecretStr) -> SecretStr:
    return SecretStr(
        hashlib.sha256(
            (token.get_secret_value() + salt.get_secret_value()).encode()
        ).hexdigest()
    )

invite async 

invite() -> SecretStr

Issue an anonymous code that admits one new account.

Source code in octomate/managers/auth.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def invite(self) -> SecretStr:
    """Issue an anonymous code that admits one new account."""
    token = SecretStr(secrets.token_urlsafe(32))
    async with async_session() as session:
        session.add(
            UserInvitation(
                token_hash=SecretStr(
                    hashlib.sha256(token.get_secret_value().encode()).hexdigest()
                ),
                expires_at=datetime.now(UTC) + self.config.invitation_lifetime,
            )
        )
        await session.commit()
        return token

register async 

register(
    username: str,
    password: SecretStr,
    invitation: SecretStr,
    *,
    name: str,
) -> User
Source code in octomate/managers/auth.py 
 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
126
127
128
129
130
131
132
133
134
135
136
async def register(
    self, username: str, password: SecretStr, invitation: SecretStr, *, name: str
) -> User:
    if not username or username != username.strip() or len(username) > 100:
        raise ValueError(
            "Username must contain 1-100 characters without surrounding spaces"
        )
    password = NewPasswordAdapter.validate_python(password)
    if not 1 <= len(name) <= 100:
        raise ValueError("Display name must contain 1-100 characters")
    async with self.lock(username), async_session() as session:
        token_hash = SecretStr(
            hashlib.sha256(invitation.get_secret_value().encode()).hexdigest()
        )
        current = await session.one_or_none(
            UserInvitation,
            expressions=[
                UserInvitation["token_hash"] == token_hash,
                UserInvitation["consumed_at"].is_(None),
                UserInvitation["expires_at"] > datetime.now(UTC),
            ],
        )
        if current is None:
            raise InvalidCredentials
        if (
            await session.one_or_none(
                User, expressions=[User["username"] == username]
            )
            is not None
        ):
            raise UsernameUnavailable
        user = User(
            username=username,
            name=name,
            password_hash=await self.hash_password(password),
        )
        session.add(user)
        now = datetime.now(UTC)
        if current.expires_at <= now:
            raise InvalidCredentials
        current.consumed_at = now
        try:
            await session.flush()
        except StaleDataError as error:
            raise InvalidCredentials from error
        await session.commit()
        return user

login async 

login(username: str, password: SecretStr) -> SessionTokens
Source code in octomate/managers/auth.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
async def login(self, username: str, password: SecretStr) -> SessionTokens:
    async with async_session() as session:
        user = await session.one_or_none(
            User, expressions=[User["username"] == username]
        )
        password_hash = (
            user.password_hash.get_secret_value()
            if user is not None and user.password_hash is not None
            else self.dummy_password_hash
        )
        try:
            await to_thread.run_sync(
                self.password_hasher.verify,
                password_hash,
                password.get_secret_value(),
                limiter=self.password_limiter,
            )
        except VerifyMismatchError as error:
            raise InvalidCredentials from error
        if user is None or user.password_hash is None:
            raise InvalidCredentials
        if self.password_hasher.check_needs_rehash(password_hash):
            user.password_hash = await self.hash_password(password)

        now = datetime.now(UTC)
        tokens = SessionTokens(
            session_id=uuid7(),
            user_id=user.id,
            access_token=SecretStr(secrets.token_urlsafe(32)),
            refresh_token=SecretStr(secrets.token_urlsafe(32)),
            access_expires_at=now + self.config.access_token_lifetime,
            refresh_expires_at=now + self.config.session_lifetime,
        )
        session.add(
            UserSession(
                id=tokens.session_id,
                user_id=user.id,
                access_token_hash=self.hash_token(
                    tokens.access_token, self.config.access_token_salt
                ),
                refresh_token_hash=self.hash_token(
                    tokens.refresh_token, self.config.refresh_token_salt
                ),
                access_expires_at=tokens.access_expires_at,
                refresh_expires_at=tokens.refresh_expires_at,
            )
        )
        await session.commit()
        return tokens

authenticate_session async 

authenticate_session(
    token: SecretStr,
) -> UserSession | None
Source code in octomate/managers/auth.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
async def authenticate_session(self, token: SecretStr) -> UserSession | None:
    now = datetime.now(UTC)
    async with async_session() as session:
        return await session.one_or_none(
            UserSession,
            expressions=[
                UserSession["access_token_hash"]
                == self.hash_token(token, self.config.access_token_salt),
                UserSession["revoked_at"].is_(None),
                UserSession["access_expires_at"] > now,
                UserSession["refresh_expires_at"] > now,
            ],
        )

set_password async 

set_password(
    username: str,
    password: SecretStr,
    *,
    current_password: SecretStr | None = None,
) -> None

Replace a password and end browser sessions; HTTP callers must supply the old password.

Source code in octomate/managers/auth.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
async def set_password(
    self,
    username: str,
    password: SecretStr,
    *,
    current_password: SecretStr | None = None,
) -> None:
    """Replace a password and end browser sessions; HTTP callers must supply the old password."""
    password = NewPasswordAdapter.validate_python(password)
    async with self.lock(username), async_session() as session:
        user = await session.one_or_none(
            User, expressions=[User["username"] == username]
        )
        if user is None or user.password_hash is None:
            raise InvalidCredentials
        if current_password is not None:
            try:
                await to_thread.run_sync(
                    self.password_hasher.verify,
                    user.password_hash.get_secret_value(),
                    current_password.get_secret_value(),
                    limiter=self.password_limiter,
                )
            except VerifyMismatchError as error:
                raise InvalidCredentials from error
        user.password_hash = await self.hash_password(password)
        now = datetime.now(UTC)
        for current in await session.list(
            UserSession,
            expressions=[
                UserSession["user_id"] == user.id,
                UserSession["revoked_at"].is_(None),
            ],
            limit=None,
        ):
            current.revoked_at = now
        await session.commit()

logout async 

logout(
    access_token: SecretStr | None,
    refresh_token: SecretStr | None,
) -> None
Source code in octomate/managers/auth.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
async def logout(
    self, access_token: SecretStr | None, refresh_token: SecretStr | None
) -> None:
    expressions = []
    if access_token is not None:
        expressions.append(
            UserSession["access_token_hash"]
            == self.hash_token(access_token, self.config.access_token_salt)
        )
    if refresh_token is not None:
        expressions.append(
            UserSession["refresh_token_hash"]
            == self.hash_token(refresh_token, self.config.refresh_token_salt)
        )
    if not expressions:
        return
    async with async_session() as session:
        for current in await session.list(
            UserSession,
            expressions=[or_(*expressions), UserSession["revoked_at"].is_(None)],
            limit=None,
        ):
            current.revoked_at = datetime.now(UTC)
        await session.commit()

refresh_session async 

refresh_session(token: SecretStr) -> SessionTokens
Source code in octomate/managers/auth.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
async def refresh_session(self, token: SecretStr) -> SessionTokens:
    now = datetime.now(UTC)
    async with async_session() as session:
        current = await session.one_or_none(
            UserSession,
            expressions=[
                UserSession["refresh_token_hash"]
                == self.hash_token(token, self.config.refresh_token_salt),
                UserSession["revoked_at"].is_(None),
                UserSession["refresh_expires_at"] > now,
            ],
        )
        if current is None:
            raise InvalidCredentials
        deadline = current.refresh_expires_at
        tokens = SessionTokens(
            session_id=current.id,
            user_id=current.user_id,
            access_token=SecretStr(secrets.token_urlsafe(32)),
            refresh_token=SecretStr(secrets.token_urlsafe(32)),
            access_expires_at=min(
                now + self.config.access_token_lifetime, deadline
            ),
            refresh_expires_at=deadline,
        )
        current.access_token_hash = self.hash_token(
            tokens.access_token, self.config.access_token_salt
        )
        current.refresh_token_hash = self.hash_token(
            tokens.refresh_token, self.config.refresh_token_salt
        )
        current.access_expires_at = tokens.access_expires_at
        try:
            await session.flush()
        except StaleDataError as error:
            raise InvalidCredentials from error
        await session.commit()
        return tokens

revoke_session async 

revoke_session(user_id: UUID7, session_id: UUID7) -> None
Source code in octomate/managers/auth.py 
304
305
306
307
308
309
310
311
312
313
314
315
async def revoke_session(self, user_id: UUID7, session_id: UUID7) -> None:
    async with async_session() as session:
        current = await session.get(UserSession, session_id)
        if current is None or current.user_id != user_id:
            raise InvalidCredentials
        if current.revoked_at is None:
            current.revoked_at = datetime.now(UTC)
            try:
                await session.flush()
            except StaleDataError as error:
                raise InvalidCredentials from error
            await session.commit()

create_api_key async 

create_api_key(
    user_id: UUID7,
    *,
    name: str,
    scopes: list[ApiKeyScope],
    expires_at: AwareDatetime | None = None,
) -> IssuedApiKey

Issue for a verified user; the HTTP caller must establish that identity.

Source code in octomate/managers/auth.py 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
async def create_api_key(
    self,
    user_id: UUID7,
    *,
    name: str,
    scopes: list[ApiKeyScope],
    expires_at: AwareDatetime | None = None,
) -> IssuedApiKey:
    """Issue for a verified user; the HTTP caller must establish that identity."""
    token = SecretStr(self.config.api_key_prefix + secrets.token_urlsafe(32))
    key = UserApiKey(
        user_id=user_id,
        name=name,
        scopes=scopes,
        expires_at=expires_at,
        key_prefix=token.get_secret_value()[:12],
        key_hash=self.hash_token(token, self.config.api_key_salt),
    )
    if key.expires_at is not None and key.expires_at <= datetime.now(UTC):
        raise ValueError("API key expiry must be in the future")
    async with async_session() as session:
        if await session.get(User, user_id) is None:
            raise InvalidCredentials
        session.add(key)
        await session.commit()
    return IssuedApiKey(key=key, token=token)

authenticate_api_key async 

authenticate_api_key(
    token: SecretStr, *, scope: ApiKeyScope
) -> UserApiKey | None
Source code in octomate/managers/auth.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
async def authenticate_api_key(
    self, token: SecretStr, *, scope: ApiKeyScope
) -> UserApiKey | None:
    async with async_session() as session:
        key = await session.one_or_none(
            UserApiKey,
            expressions=[
                UserApiKey["key_hash"]
                == self.hash_token(token, self.config.api_key_salt),
                UserApiKey["revoked_at"].is_(None),
                UserApiKey["expires_at"].is_(None)
                | (UserApiKey["expires_at"] > datetime.now(UTC)),
            ],
        )
    return key if key is not None and scope in key.scopes else None

list_api_keys async 

list_api_keys(user_id: UUID7) -> list[UserApiKey]
Source code in octomate/managers/auth.py 
360
361
362
363
364
365
366
367
368
async def list_api_keys(self, user_id: UUID7) -> list[UserApiKey]:
    async with async_session() as session:
        return list(
            await session.list(
                UserApiKey,
                expressions=[UserApiKey["user_id"] == user_id],
                limit=None,
            )
        )

revoke_api_key async 

revoke_api_key(user_id: UUID7, key_id: UUID7) -> None
Source code in octomate/managers/auth.py 
370
371
372
373
374
375
376
377
async def revoke_api_key(self, user_id: UUID7, key_id: UUID7) -> None:
    async with async_session() as session:
        key = await session.get(UserApiKey, key_id)
        if key is None or key.user_id != user_id:
            raise InvalidCredentials
        if key.revoked_at is None:
            key.revoked_at = datetime.now(UTC)
            await session.commit()