725 lines
23 KiB
Python
725 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from datetime import datetime, timedelta, timezone
|
|
import hashlib
|
|
from html import escape
|
|
import json
|
|
import os
|
|
from secrets import token_urlsafe
|
|
from typing import Any
|
|
from urllib.parse import urlencode
|
|
|
|
import httpx
|
|
from fastapi import Depends, FastAPI, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from sqlalchemy import select
|
|
|
|
from services.shared.core import Role, new_id, utc_now_iso
|
|
from services.shared.db import get_session
|
|
from services.shared.models import HealthResponse, LoginRequest, LoginResponse, UserCreate, UserOut, UserUpdate
|
|
from services.shared.security import get_actor, issue_app_token
|
|
from services.shared.sql_init import init_sql_schema
|
|
from services.shared.sql_models import AuthExternalIdentity, AuthOIDCState, AuthUser
|
|
|
|
app = FastAPI(title="auth-service", version="1.0.0")
|
|
|
|
init_sql_schema()
|
|
|
|
|
|
def _seed_users() -> None:
|
|
session = get_session()
|
|
try:
|
|
exists = session.execute(select(AuthUser.id).limit(1)).first()
|
|
if exists:
|
|
return
|
|
|
|
seeds = [
|
|
("admin", "admin123", "System Admin", Role.ADMIN),
|
|
("supervisor", "sup12345", "Shift Supervisor", Role.SUPERVISOR),
|
|
("operator", "op12345", "Call Operator", Role.OPERATOR),
|
|
("analyst", "an12345", "Business Analyst", Role.ANALYST),
|
|
]
|
|
for username, password, full_name, role in seeds:
|
|
session.add(
|
|
AuthUser(
|
|
user_id=new_id("usr"),
|
|
username=username,
|
|
password=password,
|
|
full_name=full_name,
|
|
role=role.value,
|
|
created_at=utc_now_iso(),
|
|
)
|
|
)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _bool_env(name: str, default: bool) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _oidc_enabled() -> bool:
|
|
return _bool_env("OIDC_ENABLED", False)
|
|
|
|
|
|
def _oidc_test_mode() -> bool:
|
|
return _bool_env("OIDC_TEST_MODE", False)
|
|
|
|
|
|
def _oidc_provider() -> str:
|
|
return os.getenv("OIDC_PROVIDER", "keycloak").strip() or "keycloak"
|
|
|
|
|
|
def _oidc_issuer_url() -> str:
|
|
return os.getenv("OIDC_ISSUER_URL", "").strip()
|
|
|
|
|
|
def _oidc_redirect_uri() -> str:
|
|
return os.getenv("OIDC_REDIRECT_URI", "http://localhost:8080/proxy/auth/auth/oidc/callback").strip()
|
|
|
|
|
|
def _oidc_client_id() -> str:
|
|
return os.getenv("OIDC_CLIENT_ID", "mvp-contact-center").strip()
|
|
|
|
|
|
def _oidc_client_secret() -> str:
|
|
return os.getenv("OIDC_CLIENT_SECRET", "").strip()
|
|
|
|
|
|
def _oidc_scopes() -> str:
|
|
return os.getenv("OIDC_SCOPES", "openid profile email").strip()
|
|
|
|
|
|
def _oidc_role_claim() -> str:
|
|
return os.getenv("OIDC_ROLE_CLAIM", "groups").strip() or "groups"
|
|
|
|
|
|
def _oidc_role_map() -> dict[str, str]:
|
|
default_map = {
|
|
"kc_admin": "admin",
|
|
"kc_supervisor": "supervisor",
|
|
"kc_operator": "operator",
|
|
"kc_analyst": "analyst",
|
|
}
|
|
raw = os.getenv("OIDC_ROLE_MAP_JSON", "").strip()
|
|
if not raw:
|
|
return default_map
|
|
try:
|
|
payload = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return default_map
|
|
if not isinstance(payload, dict):
|
|
return default_map
|
|
parsed: dict[str, str] = {}
|
|
for key, value in payload.items():
|
|
key_s = str(key).strip()
|
|
value_s = str(value).strip().lower()
|
|
if key_s and value_s in {role.value for role in Role}:
|
|
parsed[key_s] = value_s
|
|
return parsed or default_map
|
|
|
|
|
|
def _oidc_authorize_url_override() -> str:
|
|
return os.getenv("OIDC_AUTHORIZE_URL", "").strip()
|
|
|
|
|
|
def _oidc_token_url_override() -> str:
|
|
return os.getenv("OIDC_TOKEN_URL", "").strip()
|
|
|
|
|
|
def _oidc_jwks_url_override() -> str:
|
|
return os.getenv("OIDC_JWKS_URL", "").strip()
|
|
|
|
|
|
def _oidc_state_ttl_seconds() -> int:
|
|
return 600
|
|
|
|
|
|
def _provider_metadata() -> dict[str, Any]:
|
|
authorize_override = _oidc_authorize_url_override()
|
|
token_override = _oidc_token_url_override()
|
|
jwks_override = _oidc_jwks_url_override()
|
|
if authorize_override and token_override:
|
|
return {
|
|
"issuer": _oidc_issuer_url() or "override",
|
|
"authorization_endpoint": authorize_override,
|
|
"token_endpoint": token_override,
|
|
"jwks_uri": jwks_override,
|
|
}
|
|
|
|
issuer = _oidc_issuer_url()
|
|
if not issuer:
|
|
raise HTTPException(status_code=503, detail="OIDC issuer is not configured")
|
|
|
|
well_known = issuer.rstrip("/") + "/.well-known/openid-configuration"
|
|
try:
|
|
response = httpx.get(well_known, timeout=5)
|
|
response.raise_for_status()
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(status_code=503, detail="OIDC provider metadata unavailable") from exc
|
|
|
|
payload = response.json()
|
|
if not isinstance(payload, dict):
|
|
raise HTTPException(status_code=503, detail="OIDC provider metadata invalid")
|
|
return payload
|
|
|
|
|
|
def _provider_jwks() -> dict[str, Any]:
|
|
jwks_url = _oidc_jwks_url_override()
|
|
if not jwks_url:
|
|
jwks_url = str(_provider_metadata().get("jwks_uri") or "").strip()
|
|
if not jwks_url:
|
|
raise HTTPException(status_code=503, detail="OIDC JWKS URI unavailable")
|
|
|
|
try:
|
|
response = httpx.get(jwks_url, timeout=5)
|
|
response.raise_for_status()
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(status_code=503, detail="OIDC JWKS unavailable") from exc
|
|
|
|
payload = response.json()
|
|
if not isinstance(payload, dict):
|
|
raise HTTPException(status_code=503, detail="OIDC JWKS payload invalid")
|
|
return payload
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _iso(dt: datetime) -> str:
|
|
return dt.replace(microsecond=0).isoformat()
|
|
|
|
|
|
def _build_pkce_verifier() -> str:
|
|
return token_urlsafe(48)
|
|
|
|
|
|
def _build_pkce_challenge(code_verifier: str) -> str:
|
|
digest = hashlib.sha256(code_verifier.encode("utf-8")).digest()
|
|
return base64.urlsafe_b64encode(digest).decode("utf-8").rstrip("=")
|
|
|
|
|
|
def _html_bridge(payload: dict[str, Any], status_code: int = 200) -> HTMLResponse:
|
|
body = json.dumps(payload, ensure_ascii=False)
|
|
html = f"""<!doctype html>
|
|
<html lang="en">
|
|
<head><meta charset="utf-8"><title>OIDC Login Bridge</title></head>
|
|
<body>
|
|
<script>
|
|
(function() {{
|
|
var payload = {body};
|
|
if (window.opener) {{
|
|
window.opener.postMessage(payload, "*");
|
|
}}
|
|
window.close();
|
|
}})();
|
|
</script>
|
|
<p>{escape(payload.get("message", "Authentication completed."))}</p>
|
|
</body>
|
|
</html>"""
|
|
return HTMLResponse(content=html, status_code=status_code)
|
|
|
|
|
|
def _read_oidc_state(session_state: str) -> AuthOIDCState:
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(select(AuthOIDCState).where(AuthOIDCState.state == session_state)).scalar_one_or_none()
|
|
if not row:
|
|
raise HTTPException(status_code=400, detail="Unknown OIDC state")
|
|
if row.consumed_at:
|
|
raise HTTPException(status_code=400, detail="OIDC state already consumed")
|
|
expires = datetime.fromisoformat(row.expires_at.replace("Z", "+00:00"))
|
|
if expires < _now():
|
|
raise HTTPException(status_code=400, detail="OIDC state expired")
|
|
return row
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _consume_oidc_state(state_value: str) -> None:
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(select(AuthOIDCState).where(AuthOIDCState.state == state_value)).scalar_one_or_none()
|
|
if not row:
|
|
return
|
|
row.consumed_at = utc_now_iso()
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _save_oidc_state(state_value: str, nonce: str, code_verifier: str, redirect_uri: str, return_mode: str) -> None:
|
|
session = get_session()
|
|
try:
|
|
now = _now()
|
|
row = AuthOIDCState(
|
|
state=state_value,
|
|
nonce=nonce,
|
|
code_verifier=code_verifier,
|
|
redirect_uri=redirect_uri,
|
|
return_mode=return_mode,
|
|
created_at=_iso(now),
|
|
expires_at=_iso(now + timedelta(seconds=_oidc_state_ttl_seconds())),
|
|
consumed_at=None,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _extract_claims_from_compact_jwt(token: str) -> dict[str, Any]:
|
|
try:
|
|
parts = token.split(".")
|
|
if len(parts) != 3:
|
|
raise ValueError("compact jwt expected")
|
|
payload_part = parts[1]
|
|
padding = "=" * (-len(payload_part) % 4)
|
|
raw = base64.urlsafe_b64decode((payload_part + padding).encode("utf-8"))
|
|
payload = json.loads(raw.decode("utf-8"))
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(status_code=502, detail="OIDC id_token payload invalid") from exc
|
|
if not isinstance(payload, dict):
|
|
raise HTTPException(status_code=502, detail="OIDC id_token payload invalid")
|
|
return payload
|
|
|
|
|
|
def _map_role_from_claims(claims: dict[str, Any]) -> Role:
|
|
role_claim_name = _oidc_role_claim()
|
|
role_map = _oidc_role_map()
|
|
source = claims.get(role_claim_name)
|
|
|
|
candidates: list[str] = []
|
|
if isinstance(source, list):
|
|
candidates = [str(item) for item in source]
|
|
elif isinstance(source, str):
|
|
candidates = [source]
|
|
elif source is not None:
|
|
candidates = [str(source)]
|
|
|
|
for candidate in candidates:
|
|
mapped = role_map.get(candidate)
|
|
if mapped:
|
|
return Role(mapped)
|
|
|
|
raise HTTPException(status_code=403, detail="OIDC role is not mapped")
|
|
|
|
|
|
def _upsert_external_identity(claims: dict[str, Any], role: Role) -> None:
|
|
provider = _oidc_provider()
|
|
subject = str(claims.get("sub") or "").strip()
|
|
if not subject:
|
|
raise HTTPException(status_code=403, detail="OIDC subject missing")
|
|
username = str(
|
|
claims.get("preferred_username")
|
|
or claims.get("username")
|
|
or claims.get("email")
|
|
or subject
|
|
).strip()
|
|
now = utc_now_iso()
|
|
full_name = str(claims.get("name") or "").strip() or None
|
|
email = str(claims.get("email") or "").strip() or None
|
|
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(
|
|
select(AuthExternalIdentity).where(
|
|
AuthExternalIdentity.provider == provider,
|
|
AuthExternalIdentity.external_subject == subject,
|
|
)
|
|
).scalar_one_or_none()
|
|
if not row:
|
|
row = AuthExternalIdentity(
|
|
provider=provider,
|
|
external_subject=subject,
|
|
username=username,
|
|
email=email,
|
|
full_name=full_name,
|
|
role=role.value,
|
|
linked_user_id=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
last_login_at=now,
|
|
)
|
|
session.add(row)
|
|
else:
|
|
row.username = username
|
|
row.email = email
|
|
row.full_name = full_name
|
|
row.role = role.value
|
|
row.updated_at = now
|
|
row.last_login_at = now
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _oidc_payload_from_mock(request: Request, state_row: AuthOIDCState) -> dict[str, Any]:
|
|
if not _oidc_test_mode():
|
|
raise HTTPException(status_code=502, detail="OIDC provider exchange requires a live provider")
|
|
|
|
params = request.query_params
|
|
subject = params.get("mock_sub") or "oidc-user"
|
|
username = params.get("mock_username") or "oidc.user"
|
|
email = params.get("mock_email") or "oidc.user@example.com"
|
|
full_name = params.get("mock_full_name") or "OIDC User"
|
|
groups = [item.strip() for item in (params.get("mock_groups") or "kc_operator").split(",") if item.strip()]
|
|
claims = {
|
|
"sub": subject,
|
|
"preferred_username": username,
|
|
"email": email,
|
|
"name": full_name,
|
|
_oidc_role_claim(): groups,
|
|
"iss": _oidc_issuer_url() or "test-issuer",
|
|
"nonce": state_row.nonce,
|
|
}
|
|
return claims
|
|
|
|
|
|
def _oidc_payload_from_token_exchange(code: str, state_row: AuthOIDCState) -> dict[str, Any]:
|
|
metadata = _provider_metadata()
|
|
token_endpoint = str(metadata.get("token_endpoint") or "").strip()
|
|
if not token_endpoint:
|
|
raise HTTPException(status_code=503, detail="OIDC token endpoint unavailable")
|
|
|
|
form = {
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": state_row.redirect_uri,
|
|
"client_id": _oidc_client_id(),
|
|
"code_verifier": state_row.code_verifier,
|
|
}
|
|
if _oidc_client_secret():
|
|
form["client_secret"] = _oidc_client_secret()
|
|
|
|
try:
|
|
response = httpx.post(token_endpoint, data=form, timeout=8)
|
|
response.raise_for_status()
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(status_code=502, detail="OIDC token exchange failed") from exc
|
|
|
|
payload = response.json()
|
|
if not isinstance(payload, dict):
|
|
raise HTTPException(status_code=502, detail="OIDC token payload invalid")
|
|
|
|
claims: dict[str, Any] | None = None
|
|
if isinstance(payload.get("id_token_claims"), dict):
|
|
claims = payload["id_token_claims"]
|
|
elif isinstance(payload.get("userinfo"), dict):
|
|
claims = payload["userinfo"]
|
|
elif payload.get("id_token"):
|
|
claims = _extract_claims_from_compact_jwt(str(payload["id_token"]))
|
|
|
|
if not claims:
|
|
raise HTTPException(status_code=502, detail="OIDC claims unavailable")
|
|
|
|
issuer = str(metadata.get("issuer") or "").strip()
|
|
if issuer and claims.get("iss") and str(claims.get("iss")) != issuer:
|
|
raise HTTPException(status_code=403, detail="OIDC issuer mismatch")
|
|
if claims.get("nonce") and str(claims.get("nonce")) != state_row.nonce:
|
|
raise HTTPException(status_code=403, detail="OIDC nonce mismatch")
|
|
return claims
|
|
|
|
|
|
def _issue_login_response(user: AuthUser) -> LoginResponse:
|
|
token = issue_app_token(
|
|
subject=user.user_id,
|
|
username=user.username,
|
|
role=user.role,
|
|
auth_source="local",
|
|
full_name=user.full_name,
|
|
)
|
|
return LoginResponse(
|
|
access_token=token,
|
|
role=Role(user.role),
|
|
auth_source="local",
|
|
full_name=user.full_name,
|
|
)
|
|
|
|
|
|
def _user_out(user: AuthUser) -> UserOut:
|
|
return UserOut(
|
|
user_id=user.user_id,
|
|
username=user.username,
|
|
full_name=user.full_name,
|
|
role=Role(user.role),
|
|
)
|
|
|
|
|
|
_seed_users()
|
|
|
|
|
|
@app.get("/health", response_model=HealthResponse)
|
|
def health() -> HealthResponse:
|
|
return HealthResponse(status="ok", service="auth-service")
|
|
|
|
|
|
@app.get("/auth/roles")
|
|
def roles() -> dict:
|
|
return {"roles": [r.value for r in Role]}
|
|
|
|
|
|
@app.get("/auth/oidc/config")
|
|
def oidc_config() -> dict:
|
|
return {
|
|
"enabled": _oidc_enabled(),
|
|
"provider_label": "Keycloak",
|
|
"login_path": "/auth/oidc/start?return_mode=popup",
|
|
"auth_source": "oidc",
|
|
}
|
|
|
|
|
|
@app.get("/auth/oidc/health")
|
|
def oidc_health() -> dict:
|
|
if not _oidc_enabled():
|
|
return {
|
|
"status": "disabled",
|
|
"provider": _oidc_provider(),
|
|
"issuer": _oidc_issuer_url() or None,
|
|
"jwks_loaded": False,
|
|
}
|
|
|
|
metadata = _provider_metadata()
|
|
jwks_loaded = False
|
|
try:
|
|
_provider_jwks()
|
|
jwks_loaded = True
|
|
except HTTPException:
|
|
jwks_loaded = False
|
|
|
|
return {
|
|
"status": "ok" if jwks_loaded else "degraded",
|
|
"provider": _oidc_provider(),
|
|
"issuer": metadata.get("issuer") or _oidc_issuer_url() or None,
|
|
"jwks_loaded": jwks_loaded,
|
|
}
|
|
|
|
|
|
@app.get("/auth/oidc/start")
|
|
def oidc_start(return_mode: str = "popup") -> RedirectResponse:
|
|
if not _oidc_enabled():
|
|
raise HTTPException(status_code=503, detail="OIDC is disabled")
|
|
|
|
metadata = _provider_metadata()
|
|
authorize_url = str(metadata.get("authorization_endpoint") or "").strip()
|
|
if not authorize_url:
|
|
raise HTTPException(status_code=503, detail="OIDC authorization endpoint unavailable")
|
|
|
|
state_value = token_urlsafe(24)
|
|
nonce = token_urlsafe(16)
|
|
code_verifier = _build_pkce_verifier()
|
|
redirect_uri = _oidc_redirect_uri()
|
|
_save_oidc_state(state_value, nonce, code_verifier, redirect_uri, return_mode)
|
|
|
|
params = {
|
|
"client_id": _oidc_client_id(),
|
|
"redirect_uri": redirect_uri,
|
|
"response_type": "code",
|
|
"scope": _oidc_scopes(),
|
|
"state": state_value,
|
|
"nonce": nonce,
|
|
"code_challenge": _build_pkce_challenge(code_verifier),
|
|
"code_challenge_method": "S256",
|
|
}
|
|
return RedirectResponse(url=f"{authorize_url}?{urlencode(params)}", status_code=302)
|
|
|
|
|
|
@app.get("/auth/oidc/callback")
|
|
def oidc_callback(request: Request, state: str = "", code: str = "", error: str = "") -> HTMLResponse:
|
|
if not _oidc_enabled():
|
|
return _html_bridge({"type": "oidc-error", "message": "Corporate login is disabled"}, status_code=503)
|
|
if not state:
|
|
return _html_bridge({"type": "oidc-error", "message": "OIDC state is required"}, status_code=400)
|
|
|
|
try:
|
|
state_row = _read_oidc_state(state)
|
|
except HTTPException as exc:
|
|
return _html_bridge({"type": "oidc-error", "message": exc.detail}, status_code=exc.status_code)
|
|
|
|
if error:
|
|
_consume_oidc_state(state)
|
|
return _html_bridge({"type": "oidc-error", "message": f"OIDC provider error: {error}"}, status_code=400)
|
|
|
|
try:
|
|
if request.query_params.get("mock_sub"):
|
|
claims = _oidc_payload_from_mock(request, state_row)
|
|
else:
|
|
if not code:
|
|
raise HTTPException(status_code=400, detail="OIDC code is required")
|
|
claims = _oidc_payload_from_token_exchange(code, state_row)
|
|
|
|
role = _map_role_from_claims(claims)
|
|
_upsert_external_identity(claims, role)
|
|
|
|
username = str(
|
|
claims.get("preferred_username")
|
|
or claims.get("username")
|
|
or claims.get("email")
|
|
or claims.get("sub")
|
|
or "oidc-user"
|
|
).strip()
|
|
full_name = str(claims.get("name") or "").strip() or None
|
|
email = str(claims.get("email") or "").strip() or None
|
|
tenant_id = str(
|
|
claims.get("tenant_id")
|
|
or claims.get("organization_id")
|
|
or claims.get("org_id")
|
|
or claims.get("tid")
|
|
or ""
|
|
).strip() or None
|
|
|
|
token = issue_app_token(
|
|
subject=str(claims.get("sub") or ""),
|
|
username=username,
|
|
role=role.value,
|
|
auth_source="oidc",
|
|
provider=_oidc_provider(),
|
|
full_name=full_name,
|
|
email=email,
|
|
tenant_id=tenant_id,
|
|
)
|
|
_consume_oidc_state(state)
|
|
return _html_bridge(
|
|
{
|
|
"type": "oidc-login",
|
|
"message": "Corporate login completed",
|
|
"access_token": token,
|
|
"role": role.value,
|
|
"username": username,
|
|
"full_name": full_name,
|
|
"auth_source": "oidc",
|
|
"provider": _oidc_provider(),
|
|
"email": email,
|
|
}
|
|
)
|
|
except HTTPException as exc:
|
|
_consume_oidc_state(state)
|
|
return _html_bridge({"type": "oidc-error", "message": str(exc.detail)}, status_code=exc.status_code)
|
|
|
|
|
|
@app.post("/auth/login", response_model=LoginResponse)
|
|
def login(payload: LoginRequest) -> LoginResponse:
|
|
session = get_session()
|
|
try:
|
|
user = session.execute(
|
|
select(AuthUser).where(AuthUser.username == payload.username)
|
|
).scalar_one_or_none()
|
|
if not user or user.password != payload.password:
|
|
raise HTTPException(status_code=401, detail="Invalid credentials")
|
|
return _issue_login_response(user)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.get("/auth/me")
|
|
def me(actor: dict = Depends(get_actor)) -> dict:
|
|
return actor
|
|
|
|
|
|
@app.get("/users", response_model=list[UserOut])
|
|
def list_users(_: dict = Depends(get_actor)) -> list[UserOut]:
|
|
actor = _
|
|
if actor["role"] not in {Role.ADMIN.value, Role.SUPERVISOR.value}:
|
|
raise HTTPException(status_code=403, detail="Insufficient role")
|
|
|
|
session = get_session()
|
|
try:
|
|
users = session.execute(select(AuthUser).order_by(AuthUser.id.asc())).scalars().all()
|
|
return [_user_out(u) for u in users]
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post("/users", response_model=UserOut)
|
|
def create_user(payload: UserCreate, actor: dict = Depends(get_actor)) -> UserOut:
|
|
if actor["role"] != Role.ADMIN.value:
|
|
raise HTTPException(status_code=403, detail="Insufficient role")
|
|
|
|
session = get_session()
|
|
try:
|
|
existing = session.execute(
|
|
select(AuthUser).where(AuthUser.username == payload.username)
|
|
).scalar_one_or_none()
|
|
if existing:
|
|
raise HTTPException(status_code=409, detail="Username already exists")
|
|
|
|
user = AuthUser(
|
|
user_id=new_id("usr"),
|
|
username=payload.username,
|
|
password=payload.password,
|
|
full_name=payload.full_name,
|
|
role=payload.role.value,
|
|
created_at=utc_now_iso(),
|
|
)
|
|
session.add(user)
|
|
session.commit()
|
|
return _user_out(user)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.get("/users/{user_id}", response_model=UserOut)
|
|
def get_user(user_id: str, actor: dict = Depends(get_actor)) -> UserOut:
|
|
if actor["role"] not in {Role.ADMIN.value, Role.SUPERVISOR.value}:
|
|
raise HTTPException(status_code=403, detail="Insufficient role")
|
|
|
|
session = get_session()
|
|
try:
|
|
user = session.execute(select(AuthUser).where(AuthUser.user_id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return _user_out(user)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.patch("/users/{user_id}", response_model=UserOut)
|
|
def update_user(user_id: str, payload: UserUpdate, actor: dict = Depends(get_actor)) -> UserOut:
|
|
if actor["role"] != Role.ADMIN.value:
|
|
raise HTTPException(status_code=403, detail="Insufficient role")
|
|
|
|
session = get_session()
|
|
try:
|
|
user = session.execute(select(AuthUser).where(AuthUser.user_id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
if payload.password is not None:
|
|
user.password = payload.password
|
|
if payload.full_name is not None:
|
|
user.full_name = payload.full_name
|
|
if payload.role is not None:
|
|
user.role = payload.role.value
|
|
|
|
session.commit()
|
|
session.refresh(user)
|
|
return _user_out(user)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.delete("/users/{user_id}")
|
|
def delete_user(user_id: str, actor: dict = Depends(get_actor)) -> dict:
|
|
if actor["role"] != Role.ADMIN.value:
|
|
raise HTTPException(status_code=403, detail="Insufficient role")
|
|
|
|
session = get_session()
|
|
try:
|
|
user = session.execute(select(AuthUser).where(AuthUser.user_id == user_id)).scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if actor.get("user") == user.username:
|
|
raise HTTPException(status_code=409, detail="Cannot delete current user")
|
|
|
|
session.delete(user)
|
|
session.commit()
|
|
return {"user_id": user_id, "deleted": True}
|
|
finally:
|
|
session.close()
|