Show the OIDC profile picture with an identicon fallback
CI / Build and push docker image (push) Successful in 2m9s

GET /api/me now carries the profile picture URL when the provider
supplies an absolute http(s) one; the SPA shows it in the lobby navbar
and draws a deterministic identicon from the sub otherwise, falling back
to the identicon when the image fails to load.

The identicon is a 5x5 mirrored grid seeded by an inline FNV-1a hash so
it stays stable across Rust versions and is unit-testable on the host.
The dev mock IdP hands alice and bob a picture claim.
This commit is contained in:
2026-10-15 00:14:17 +08:00
parent a9eb97e6e1
commit 46f10f145a
13 changed files with 312 additions and 28 deletions
+4 -2
View File
@@ -11,7 +11,8 @@
"claims": {
"sub": "alice",
"preferred_username": "alice",
"name": "Alice"
"name": "Alice",
"picture": "https://i.pravatar.cc/128?u=alice"
}
},
{
@@ -20,7 +21,8 @@
"claims": {
"sub": "bob",
"preferred_username": "bob",
"name": "Bob"
"name": "Bob",
"picture": "https://i.pravatar.cc/128?u=bob"
}
},
{
@@ -60,6 +60,19 @@ def display_name(user: OIDCUser) -> str:
return user.sub
def picture(user: OIDCUser) -> Optional[str]:
"""The user's avatar URL, when the provider supplies a usable one.
Only absolute ``http(s)`` URLs are exported: the claim ends up in an
``<img src>`` in the SPA, where relative or script-ish schemes have no
business.
"""
value = user.picture
if isinstance(value, str) and value.startswith(("http://", "https://")):
return value
return None
def current_user(oidc: OIDCProviders, ctx: HttpContext) -> Optional[OIDCUser]:
"""Return the authenticated user of an HTTP request, if any.
@@ -126,8 +126,9 @@ ME_SCHEMA = _object(
{
"sub": {"type": "string"},
"name": {"type": "string"},
"picture": _nullable({"type": "string"}),
},
("sub", "name"),
("sub", "name", "picture"),
)
AUTH_PROVIDER_SCHEMA = _object(
@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING
from kaya.core import HttpContext, KayaApp
from kaya.openapi import operation, schema_ref
from ..auth import current_user, display_name, require_auth
from ..auth import current_user, display_name, picture, require_auth
from ..http import send_json
from ..openapi import error_response, json_response
@@ -18,7 +18,9 @@ def register(app: KayaApp, platform: "Platform") -> None:
@app.GET("/api/me")
@operation(summary="Current user",
description="Returns the authenticated user's identity from the "
"session; 401 when not logged in.",
"session, including the profile picture URL when "
"the provider supplies one; 401 when not logged "
"in.",
tags=["auth"],
responses={
200: json_response("The current user", schema_ref("Me")),
@@ -28,4 +30,12 @@ def register(app: KayaApp, platform: "Platform") -> None:
async def me(ctx: HttpContext) -> None:
user = current_user(platform.oidc, ctx)
assert user is not None # enforced by @require_auth
await send_json(ctx, 200, {"sub": user.sub, "name": display_name(user)})
await send_json(
ctx,
200,
{
"sub": user.sub,
"name": display_name(user),
"picture": picture(user),
},
)
@@ -19,6 +19,7 @@ from tavolo.platform.auth import (
current_user,
display_name,
get_ws_user,
picture,
require_auth,
session_user,
)
@@ -121,6 +122,23 @@ class DisplayNameTest(unittest.TestCase):
self.assertEqual("alice", display_name(user))
class PictureTest(unittest.TestCase):
"""Only absolute http(s) picture claims are exported to the SPA."""
def test_uses_an_absolute_http_url(self) -> None:
user = OIDCUser({"sub": "alice", "picture": "https://cdn.example.com/a.png"})
self.assertEqual("https://cdn.example.com/a.png", picture(user))
def test_rejects_relative_and_script_schemes(self) -> None:
for value in ("/relative.png", "javascript:alert(1)", "data:image/png;base64,x"):
with self.subTest(picture=value):
user = OIDCUser({"sub": "alice", "picture": value})
self.assertIsNone(picture(user))
def test_missing_claim_is_none(self) -> None:
self.assertIsNone(picture(OIDCUser({"sub": "alice"})))
class CurrentUserTest(unittest.TestCase):
def test_anonymous_session_has_no_user(self) -> None:
self.assertIsNone(current_user(_providers("default"), _ctx({})))
@@ -310,7 +310,10 @@ class MeRouteTest(unittest.TestCase):
with oidc_user(platform.oidc, "alice"):
response = await client.get("/api/me")
self.assertEqual(200, response.status_code)
self.assertEqual({"sub": "alice", "name": "alice"}, response.json())
self.assertEqual(
{"sub": "alice", "name": "alice", "picture": None},
response.json(),
)
@async_test
async def test_me_unauthenticated(self) -> None:
+13 -4
View File
@@ -24,25 +24,34 @@ _PROVIDER = next(iter(oidc.providers))
def make_user(
sub: str, name: Optional[str] = None, email: Optional[str] = None
sub: str,
name: Optional[str] = None,
email: Optional[str] = None,
picture: Optional[str] = None,
) -> OIDCUser:
claims = {"sub": sub, "preferred_username": name or sub}
if email:
claims["email"] = email
if picture:
claims["picture"] = picture
return OIDCUser(claims)
@contextlib.contextmanager
def oidc_user(
sub: str, name: Optional[str] = None, email: Optional[str] = None
sub: str,
name: Optional[str] = None,
email: Optional[str] = None,
picture: Optional[str] = None,
) -> Iterator[OIDCUser]:
"""Context manager: make ``oidc.get_users`` return this user.
``email`` is optional so tests can simulate an IdP that still returns
the claim: the app requests only the ``openid`` and ``profile`` scopes
and ignores email where present.
and ignores email where present. ``picture`` simulates the profile
scope's avatar claim.
"""
user = make_user(sub, name, email)
user = make_user(sub, name, email, picture)
patcher = _mock.patch.object(oidc, "get_users", return_value={_PROVIDER: user})
patcher.start()
try:
+27 -2
View File
@@ -22,8 +22,33 @@ class MeRouteTest(unittest.TestCase):
with oidc_user("alice", email="alice@example.com"):
response = await client.get("/api/me")
self.assertEqual(200, response.status_code)
# The email claim the IdP returns is neither stored nor exposed.
self.assertEqual({"sub": "alice", "name": "alice"}, response.json())
# The email claim the IdP returns is neither stored nor exposed;
# no picture claim in this request.
self.assertEqual(
{"sub": "alice", "name": "alice", "picture": None},
response.json(),
)
@async_test
async def test_me_includes_the_profile_picture(self) -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user("alice", picture="https://cdn.example.com/alice.png"):
response = await client.get("/api/me")
self.assertEqual(200, response.status_code)
self.assertEqual(
"https://cdn.example.com/alice.png",
response.json()["picture"],
)
@async_test
async def test_me_rejects_non_http_picture_claims(self) -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
with oidc_user("alice", picture="javascript:alert(1)"):
response = await client.get("/api/me")
self.assertEqual(200, response.status_code)
self.assertIsNone(response.json()["picture"])
@async_test
async def test_me_unauthenticated(self) -> None: