Implement NetKingdom identity shell for coulomb.social (CSOC-WP-0002)

Django scaffold aligned with the business delivery lane: tenant-keyed
Member model without passwords, identity app as sole OIDC/session
boundary, dev-claims login, authenticated /app/ shell, ADR-0001, and
tests. T01/T02/T05/T06 done; OIDC registration, real user-engine HTTP,
flex-auth, and packaging remain open.
This commit is contained in:
tegwick 2026-08-09 01:45:05 +02:00
parent 2ec7761504
commit 01da195c13
51 changed files with 2215 additions and 41 deletions

View file

View file

View file

@ -0,0 +1,7 @@
from django.apps import AppConfig
class CoreConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "coulomb_social.apps.core"
label = "core"

View file

@ -0,0 +1,9 @@
from django.conf import settings
def site_context(request):
return {
"site_name": "coulomb.social",
"default_tenant_id": settings.DEFAULT_TENANT_ID,
"oidc_enabled": settings.OIDC_ENABLED,
}

View file

@ -0,0 +1,10 @@
from django.urls import path
from . import views
app_name = "core"
urlpatterns = [
path("", views.landing, name="landing"),
path("app/", views.app_home, name="app_home"),
]

View file

@ -0,0 +1,7 @@
from django.urls import path
from . import views
urlpatterns = [
path("", views.healthz, name="healthz"),
]

View file

@ -0,0 +1,41 @@
from django.contrib.auth.decorators import login_required
from django.http import HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import render
def landing(request: HttpRequest) -> HttpResponse:
if request.user.is_authenticated:
from django.shortcuts import redirect
return redirect("core:app_home")
return render(request, "core/landing.html")
@login_required
def app_home(request: HttpRequest) -> HttpResponse:
member = getattr(request.user, "member", None)
return render(
request,
"core/app_home.html",
{
"member": member,
"principal": {
"username": request.user.get_username(),
"display_name": (
member.display_name
if member and member.display_name
else request.user.get_full_name() or request.user.get_username()
),
"issuer": member.issuer if member else "",
"subject": member.subject if member else "",
"user_engine_user_id": (
str(member.user_engine_user_id) if member else ""
),
"tenant_id": member.tenant_id if member else "",
},
},
)
def healthz(_request: HttpRequest) -> JsonResponse:
return JsonResponse({"status": "ok", "service": "coulomb-social"})

View file

@ -0,0 +1 @@
"""Auth boundary module — sole place that talks OIDC / sessions (ADR-0001, §2.3)."""

View file

@ -0,0 +1,7 @@
from django.apps import AppConfig
class IdentityConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "coulomb_social.apps.identity"
label = "identity"

View file

@ -0,0 +1,33 @@
"""flex-auth PEP port — fail-closed for sensitive actions when PDP missing."""
from __future__ import annotations
from dataclasses import dataclass
from django.conf import settings
@dataclass(frozen=True)
class AuthzDecision:
allow: bool
reason: str
decision_id: str = ""
def check(action: str, *, resource: str = "shell", subject: str = "") -> AuthzDecision:
"""Decide whether `action` is allowed.
- shell:view is allowed for any authenticated principal (shell smoke).
- other actions require FLEX_AUTH_BASE_URL; until wired, deny.
"""
if action == "shell:view":
return AuthzDecision(allow=True, reason="shell-view-authenticated")
if not settings.FLEX_AUTH_BASE_URL:
return AuthzDecision(
allow=False,
reason="flex-auth-not-configured-fail-closed",
)
# TODO(CSOC-WP-0002-T07): HTTP call to flex-auth PDP
return AuthzDecision(allow=False, reason="flex-auth-http-not-implemented")

View file

@ -0,0 +1,103 @@
"""OIDC/PKCE helpers for NetKingdom IAM Profile issuers."""
from __future__ import annotations
import secrets
from typing import Any
from urllib.parse import urlencode
import httpx
from authlib.integrations.httpx_client import OAuth2Client
from django.conf import settings
class OIDCConfigurationError(RuntimeError):
pass
def oidc_configured() -> bool:
return bool(
settings.OIDC_ENABLED
and settings.OIDC_ISSUER
and settings.OIDC_CLIENT_ID
and settings.OIDC_REDIRECT_URI
)
def discovery_document() -> dict[str, Any]:
if not settings.OIDC_ISSUER and not settings.OIDC_DISCOVERY_URL:
raise OIDCConfigurationError("OIDC_ISSUER or OIDC_DISCOVERY_URL required")
url = settings.OIDC_DISCOVERY_URL or (
settings.OIDC_ISSUER.rstrip("/") + "/.well-known/openid-configuration"
)
resp = httpx.get(url, timeout=15.0)
resp.raise_for_status()
return resp.json()
def build_authorization_url(*, state: str, code_verifier: str) -> str:
if not oidc_configured():
raise OIDCConfigurationError("OIDC is not enabled/configured")
doc = discovery_document()
auth_endpoint = doc["authorization_endpoint"]
client = OAuth2Client(
client_id=settings.OIDC_CLIENT_ID,
client_secret=settings.OIDC_CLIENT_SECRET or None,
redirect_uri=settings.OIDC_REDIRECT_URI,
scope=settings.OIDC_SCOPES,
code_challenge_method="S256",
)
uri, _ = client.create_authorization_url(
auth_endpoint,
state=state,
code_verifier=code_verifier,
)
return uri
def exchange_code(code: str, *, code_verifier: str) -> dict[str, Any]:
doc = discovery_document()
token_endpoint = doc["token_endpoint"]
client = OAuth2Client(
client_id=settings.OIDC_CLIENT_ID,
client_secret=settings.OIDC_CLIENT_SECRET or None,
redirect_uri=settings.OIDC_REDIRECT_URI,
)
token = client.fetch_token(
token_endpoint,
code=code,
code_verifier=code_verifier,
grant_type="authorization_code",
)
return token
def fetch_userinfo(access_token: str) -> dict[str, Any]:
doc = discovery_document()
userinfo_endpoint = doc.get("userinfo_endpoint")
if not userinfo_endpoint:
return {}
resp = httpx.get(
userinfo_endpoint,
headers={"Authorization": f"Bearer {access_token}"},
timeout=15.0,
)
resp.raise_for_status()
return resp.json()
def new_pkce_pair() -> tuple[str, str]:
"""Return (state, code_verifier). Challenge is computed by Authlib client."""
state = secrets.token_urlsafe(24)
code_verifier = secrets.token_urlsafe(48)
return state, code_verifier
def claims_from_token_response(token: dict[str, Any], userinfo: dict[str, Any]) -> dict[str, Any]:
"""Merge id_token claims (if present as dict) with userinfo."""
claims: dict[str, Any] = {}
# authlib may leave id_token as JWT string; userinfo is preferred when available
claims.update(userinfo or {})
if not claims.get("sub") and isinstance(token.get("userinfo"), dict):
claims.update(token["userinfo"])
return claims

View file

@ -0,0 +1,66 @@
"""Session establishment after verified identity claims."""
from __future__ import annotations
from django.conf import settings
from django.contrib.auth import login
from django.http import HttpRequest
from django.utils import timezone
from coulomb_social.apps.members.models import Member, User
from .user_engine import IdentityClaims, get_user_engine_client
def establish_session(request: HttpRequest, claims: IdentityClaims) -> Member:
"""Link platform identity → Member, log Django user in. No passwords."""
tenant_id = settings.DEFAULT_TENANT_ID
application_id = settings.USER_ENGINE_APPLICATION_ID
link = get_user_engine_client().link_or_create(
claims, tenant_id=tenant_id, application_id=application_id
)
member = Member.objects.filter(issuer=claims.issuer, subject=claims.subject).first()
display = claims.name or claims.preferred_username or claims.email or claims.subject
username = f"{claims.subject}@{_issuer_slug(claims.issuer)}"[:255]
if member is None:
user = User.objects.create_user(
username=username,
email=claims.email or "",
full_name=display,
)
member = Member.objects.create(
tenant_id=tenant_id,
user=user,
user_engine_user_id=link.user_id,
issuer=claims.issuer,
subject=claims.subject,
display_name=display,
email=claims.email or "",
)
else:
user = member.user
user.email = claims.email or user.email
user.full_name = display or user.full_name
user.save(update_fields=["email", "full_name"])
member.display_name = display
member.email = claims.email or member.email
member.user_engine_user_id = link.user_id
member.tenant_id = tenant_id
member.last_login_at = timezone.now()
member.save()
login(request, user, backend="django.contrib.auth.backends.ModelBackend")
return member
def _issuer_slug(issuer: str) -> str:
return (
issuer.replace("https://", "")
.replace("http://", "")
.replace("/", "_")
.replace(":", "_")[:64]
or "issuer"
)

View file

@ -0,0 +1,12 @@
from django.urls import path
from . import views
app_name = "identity"
urlpatterns = [
path("login/", views.login_start, name="login"),
path("callback/", views.oidc_callback, name="callback"),
path("logout/", views.logout_view, name="logout"),
path("dev-login/", views.dev_login, name="dev_login"),
]

View file

@ -0,0 +1,93 @@
"""user-engine integration port.
Production will call the user-engine HTTP API. Until that service is wired
for this app, an in-process stub provisions stable user ids from claims so
the shell and tests can run offline.
"""
from __future__ import annotations
import hashlib
import uuid
from dataclasses import dataclass
from typing import Protocol
from django.conf import settings
@dataclass(frozen=True)
class IdentityClaims:
issuer: str
subject: str
email: str = ""
name: str = ""
preferred_username: str = ""
@dataclass(frozen=True)
class UserEngineLink:
user_id: str
application_id: str
tenant_id: str
created: bool
class UserEngineClient(Protocol):
def link_or_create(
self,
claims: IdentityClaims,
*,
tenant_id: str,
application_id: str,
) -> UserEngineLink: ...
class StubUserEngineClient:
"""Deterministic offline user-engine stand-in (not for production identity)."""
def link_or_create(
self,
claims: IdentityClaims,
*,
tenant_id: str,
application_id: str,
) -> UserEngineLink:
digest = hashlib.sha256(
f"{claims.issuer}|{claims.subject}".encode()
).hexdigest()[:32]
# UUID-shaped stable id for readability in the shell
user_id = str(uuid.UUID(digest))
return UserEngineLink(
user_id=user_id,
application_id=application_id,
tenant_id=tenant_id,
created=True,
)
class HttpUserEngineClient:
"""Minimal HTTP client placeholder — expand when USER_ENGINE_BASE_URL is live."""
def __init__(self, base_url: str) -> None:
self.base_url = base_url.rstrip("/")
def link_or_create(
self,
claims: IdentityClaims,
*,
tenant_id: str,
application_id: str,
) -> UserEngineLink:
# Until the production contract endpoint is confirmed, fall back to stub
# semantics while recording that HTTP mode was requested.
# TODO(CSOC-WP-0002-T04): replace with real projection/link API.
return StubUserEngineClient().link_or_create(
claims, tenant_id=tenant_id, application_id=application_id
)
def get_user_engine_client() -> UserEngineClient:
base = (settings.USER_ENGINE_BASE_URL or "").strip()
if base:
return HttpUserEngineClient(base)
return StubUserEngineClient()

View file

@ -0,0 +1,118 @@
"""OIDC login/callback/logout + DEBUG dev-login. Sole auth entrypoints."""
from __future__ import annotations
import logging
from django.conf import settings
from django.contrib import messages
from django.contrib.auth import logout
from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest
from django.shortcuts import redirect, render
from django.urls import reverse
from django.views.decorators.http import require_GET, require_http_methods
from . import oidc
from .services import establish_session
from .user_engine import IdentityClaims
logger = logging.getLogger(__name__)
SESSION_OIDC_STATE = "oidc_state"
SESSION_OIDC_VERIFIER = "oidc_code_verifier"
@require_GET
def login_start(request: HttpRequest) -> HttpResponse:
if request.user.is_authenticated:
return redirect(settings.LOGIN_REDIRECT_URL)
if oidc.oidc_configured():
state, verifier = oidc.new_pkce_pair()
request.session[SESSION_OIDC_STATE] = state
request.session[SESSION_OIDC_VERIFIER] = verifier
try:
url = oidc.build_authorization_url(state=state, code_verifier=verifier)
except Exception:
logger.exception("OIDC authorization URL build failed")
messages.error(request, "Identity provider is unavailable. Try again later.")
return redirect("core:landing")
return redirect(url)
if settings.DEBUG:
return redirect("identity:dev_login")
messages.error(request, "Sign-in is not configured (OIDC_ENABLED=false).")
return redirect("core:landing")
@require_GET
def oidc_callback(request: HttpRequest) -> HttpResponse:
if not oidc.oidc_configured():
return HttpResponseBadRequest("OIDC is not enabled")
error = request.GET.get("error")
if error:
messages.error(request, f"Sign-in failed: {error}")
return redirect("core:landing")
code = request.GET.get("code")
state = request.GET.get("state")
expected_state = request.session.pop(SESSION_OIDC_STATE, None)
verifier = request.session.pop(SESSION_OIDC_VERIFIER, None)
if not code or not state or state != expected_state or not verifier:
return HttpResponseBadRequest("Invalid OIDC callback state")
try:
token = oidc.exchange_code(code, code_verifier=verifier)
userinfo = oidc.fetch_userinfo(token.get("access_token", ""))
raw = oidc.claims_from_token_response(token, userinfo)
except Exception:
logger.exception("OIDC token exchange failed")
messages.error(request, "Could not complete sign-in with the identity provider.")
return redirect("core:landing")
sub = raw.get("sub")
if not sub:
return HttpResponseBadRequest("Token missing subject")
issuer = raw.get("iss") or settings.OIDC_ISSUER
claims = IdentityClaims(
issuer=str(issuer),
subject=str(sub),
email=str(raw.get("email") or ""),
name=str(raw.get("name") or ""),
preferred_username=str(raw.get("preferred_username") or ""),
)
establish_session(request, claims)
return redirect(settings.LOGIN_REDIRECT_URL)
@require_http_methods(["GET", "POST"])
def dev_login(request: HttpRequest) -> HttpResponse:
"""Local-only claims form when OIDC is off. Never enable outside DEBUG."""
if not settings.DEBUG or settings.OIDC_ENABLED:
return HttpResponseBadRequest("Dev login only when DEBUG and OIDC disabled")
if request.method == "POST":
subject = (request.POST.get("subject") or "").strip()
if not subject:
messages.error(request, "Subject is required")
return render(request, "identity/dev_login.html")
claims = IdentityClaims(
issuer=request.POST.get("issuer") or "https://local.dev/issuer",
subject=subject,
email=(request.POST.get("email") or "").strip(),
name=(request.POST.get("name") or "").strip(),
preferred_username=(request.POST.get("preferred_username") or "").strip(),
)
establish_session(request, claims)
return redirect(settings.LOGIN_REDIRECT_URL)
return render(request, "identity/dev_login.html")
@require_http_methods(["GET", "POST"])
def logout_view(request: HttpRequest) -> HttpResponse:
logout(request)
return redirect(settings.LOGOUT_REDIRECT_URL)

View file

View file

@ -0,0 +1,39 @@
from django.contrib import admin
from django.contrib.auth.admin import UserAdmin as DjangoUserAdmin
from .models import Member, User
@admin.register(User)
class UserAdmin(DjangoUserAdmin):
ordering = ("username",)
list_display = ("username", "email", "full_name", "is_staff", "is_active")
search_fields = ("username", "email", "full_name")
fieldsets = (
(None, {"fields": ("username",)}),
("Profile", {"fields": ("full_name", "email")}),
("Permissions", {"fields": ("is_active", "is_staff", "is_superuser", "groups", "user_permissions")}),
)
add_fieldsets = (
(
None,
{
"classes": ("wide",),
"fields": ("username", "is_staff", "is_superuser"),
},
),
)
@admin.register(Member)
class MemberAdmin(admin.ModelAdmin):
list_display = (
"display_name",
"tenant_id",
"subject",
"user_engine_user_id",
"last_login_at",
)
search_fields = ("display_name", "email", "subject", "user_engine_user_id")
list_filter = ("tenant_id", "issuer")
readonly_fields = ("id", "created_at", "updated_at")

View file

@ -0,0 +1,7 @@
from django.apps import AppConfig
class MembersConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "coulomb_social.apps.members"
label = "members"

View file

@ -0,0 +1,132 @@
# Generated by Django 6.1 on 2026-08-08 23:43
import django.db.models.deletion
import django.utils.timezone
import uuid
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
("auth", "0012_alter_user_first_name_max_length"),
]
operations = [
migrations.CreateModel(
name="User",
fields=[
("password", models.CharField(max_length=128, verbose_name="password")),
(
"last_login",
models.DateTimeField(
blank=True, null=True, verbose_name="last login"
),
),
(
"is_superuser",
models.BooleanField(
default=False,
help_text="Designates that this user has all permissions without explicitly assigning them.",
verbose_name="superuser status",
),
),
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
("username", models.CharField(max_length=255, unique=True)),
("email", models.EmailField(blank=True, max_length=254)),
("full_name", models.CharField(blank=True, max_length=255)),
("is_active", models.BooleanField(default=True)),
("is_staff", models.BooleanField(default=False)),
(
"date_joined",
models.DateTimeField(default=django.utils.timezone.now),
),
(
"groups",
models.ManyToManyField(
blank=True,
help_text="The groups this user belongs to. A user will get all permissions granted to each of their groups.",
related_name="user_set",
related_query_name="user",
to="auth.group",
verbose_name="groups",
),
),
(
"user_permissions",
models.ManyToManyField(
blank=True,
help_text="Specific permissions for this user.",
related_name="user_set",
related_query_name="user",
to="auth.permission",
verbose_name="user permissions",
),
),
],
options={
"verbose_name": "user",
"verbose_name_plural": "users",
},
),
migrations.CreateModel(
name="Member",
fields=[
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
("tenant_id", models.CharField(db_index=True, max_length=64)),
("user_engine_user_id", models.CharField(db_index=True, max_length=64)),
("issuer", models.CharField(max_length=512)),
("subject", models.CharField(max_length=255)),
("display_name", models.CharField(blank=True, max_length=255)),
("email", models.EmailField(blank=True, max_length=254)),
("created_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
("last_login_at", models.DateTimeField(blank=True, null=True)),
(
"user",
models.OneToOneField(
on_delete=django.db.models.deletion.CASCADE,
related_name="member",
to=settings.AUTH_USER_MODEL,
),
),
],
options={
"indexes": [
models.Index(
fields=["tenant_id", "subject"],
name="members_mem_tenant__dd4696_idx",
)
],
"constraints": [
models.UniqueConstraint(
fields=("issuer", "subject"),
name="members_member_issuer_subject_uniq",
),
models.UniqueConstraint(
fields=("tenant_id", "user_engine_user_id"),
name="members_member_tenant_ue_user_uniq",
),
],
},
),
]

View file

@ -0,0 +1,100 @@
"""Member and User models — no local password authentication.
Identity is established via NetKingdom OIDC (IAM Profile). Passwords are never
collected or validated; Django users always have an unusable password.
"""
from __future__ import annotations
import uuid
from django.contrib.auth.models import AbstractBaseUser, BaseUserManager, PermissionsMixin
from django.db import models
from django.utils import timezone
class UserManager(BaseUserManager):
def create_user(self, username: str, **extra_fields):
if not username:
raise ValueError("username is required")
user = self.model(username=username, **extra_fields)
user.set_unusable_password()
user.save(using=self._db)
return user
def create_superuser(self, username: str, **extra_fields):
extra_fields.setdefault("is_staff", True)
extra_fields.setdefault("is_superuser", True)
return self.create_user(username, **extra_fields)
class User(AbstractBaseUser, PermissionsMixin):
"""Session principal. Credentials live at the NetKingdom issuer, not here."""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
username = models.CharField(max_length=255, unique=True)
email = models.EmailField(blank=True)
full_name = models.CharField(max_length=255, blank=True)
is_active = models.BooleanField(default=True)
is_staff = models.BooleanField(default=False)
date_joined = models.DateTimeField(default=timezone.now)
objects = UserManager()
USERNAME_FIELD = "username"
REQUIRED_FIELDS: list[str] = []
class Meta:
verbose_name = "user"
verbose_name_plural = "users"
def get_full_name(self) -> str:
return self.full_name or self.username
def get_short_name(self) -> str:
return self.full_name or self.username
def __str__(self) -> str:
return self.username
class Member(models.Model):
"""Local app context for a platform user within a tenant.
Links NetKingdom OIDC subject + user-engine user id. No password fields.
"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
tenant_id = models.CharField(max_length=64, db_index=True)
user = models.OneToOneField(
User,
on_delete=models.CASCADE,
related_name="member",
)
# Stable id from user-engine (string to avoid coupling to UUID vs ULID choice)
user_engine_user_id = models.CharField(max_length=64, db_index=True)
issuer = models.CharField(max_length=512)
subject = models.CharField(max_length=255)
display_name = models.CharField(max_length=255, blank=True)
email = models.EmailField(blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
last_login_at = models.DateTimeField(null=True, blank=True)
class Meta:
constraints = [
models.UniqueConstraint(
fields=["issuer", "subject"],
name="members_member_issuer_subject_uniq",
),
models.UniqueConstraint(
fields=["tenant_id", "user_engine_user_id"],
name="members_member_tenant_ue_user_uniq",
),
]
indexes = [
models.Index(fields=["tenant_id", "subject"]),
]
def __str__(self) -> str:
return f"{self.display_name or self.subject}@{self.tenant_id}"