update

devplacepy_services/base/auth.py

@@ -0,0 +1,61 @@
+# retoor <retoor@molodetz.nl>
+
+from __future__ import annotations
+
+import os
+from typing import Annotated
+
+from fastapi import Header, Request
+
+from devplacepy_services.base.errors import http_error
+from devplacepy_services.base.schemas import InternalUser
+
+_KEY_CACHE = ""
+
+
+def internal_gateway_key() -> str:
+    global _KEY_CACHE
+    if _KEY_CACHE:
+        return _KEY_CACHE
+    env_key = os.environ.get("DEVPLACE_GATEWAY_INTERNAL_KEY", "").strip()
+    if env_key:
+        _KEY_CACHE = env_key
+        return env_key
+    try:
+        from devplacepy.db_client import internal_gateway_key as load_key
+
+        loaded = load_key().strip()
+        if loaded:
+            _KEY_CACHE = loaded
+        return loaded
+    except ImportError:
+        return ""
+
+
+def set_internal_gateway_key(value: str) -> None:
+    global _KEY_CACHE
+    _KEY_CACHE = value.strip()
+
+
+def validate_internal_key(presented: str | None) -> bool:
+    expected = internal_gateway_key()
+    if not expected:
+        return True
+    if not presented:
+        return False
+    return presented.strip() == expected
+
+
+def require_internal_key(request: Request) -> None:
+    if validate_internal_key(request.headers.get("X-Internal-Key")):
+        return
+    raise http_error(401, "Unauthorized", "unauthorized")
+
+
+async def internal_user(
+    request: Request,
+    x_authenticated_user: Annotated[str | None, Header()] = None,
+) -> InternalUser:
+    require_internal_key(request)
+    uid = x_authenticated_user.strip() if x_authenticated_user else None
+    return InternalUser(uid=uid or None)