fix: proxy auth signing and project root resolution

docs/setup/security.md

@@ -40,6 +40,24 @@ openssl rand -base64 32
 
 ## Authentication
 
+### Phlo Reverse Proxy Authentication
+
+If you put Phlo behind a trusted reverse proxy, configure the app to accept asserted identity
+headers only from that proxy and bind those headers into a shared-secret signature:
+
+```bash
+# .phlo/.env.local
+PHLO_AUTH_PROXY_ENABLED=true
+PHLO_AUTH_PROXY_TRUSTED_PROXIES=127.0.0.1/32,10.0.0.0/8
+PHLO_AUTH_PROXY_SHARED_SECRET=<generate-strong-random-secret>
+PHLO_AUTH_PROXY_HEADER_SUBJECT=X-Remote-User
+PHLO_AUTH_PROXY_HEADER_EMAIL=X-Remote-Email
+PHLO_AUTH_PROXY_HEADER_GROUPS=X-Remote-Groups
+```
+
+When `PHLO_AUTH_PROXY_SHARED_SECRET` is set, the proxy must sign the timestamp, remote address,
+request path, and asserted identity headers so downstream header changes are rejected.
+
 ### Option 1: LDAP Authentication
 
 LDAP works with Trino and MinIO. Configure your LDAP server details:

src/phlo/capabilities/authentication.py

@@ -8,7 +8,7 @@
 import os
 import secrets
 from contextlib import suppress
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from typing import Any
 
 from phlo.capabilities.interfaces import (
@@ -208,7 +208,7 @@ def _is_session_valid(self, session: AuthenticatedSession) -> bool:
         """Check if session is still valid."""
         if session.expires_at is None:
             return True
-        return datetime.now(timezone.utc) < session.expires_at  # noqa: UP017
+        return datetime.now(UTC) < session.expires_at
 
 
 class ProxyAuthenticationProvider:
@@ -225,6 +225,7 @@ def __init__(
         header_subject: str = "X-Remote-User",
         header_email: str = "X-Remote-Email",
         header_groups: str = "X-Remote-Groups",
+        shared_secret: str | None = None,
     ):
         self._trusted_networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = []
         self._trusted_hosts: set[str] = set()
@@ -240,6 +241,22 @@ def __init__(
         self._header_subject = header_subject.lower()
         self._header_email = header_email.lower()
         self._header_groups = header_groups.lower()
+        self._signature_header = "x-phlo-proxy-signature"
+        self._timestamp_header = "x-phlo-proxy-timestamp"
+        self._shared_secret = shared_secret
+
+    def _identity_headers(self, request_context: RequestContext) -> tuple[str, str | None, str]:
+        """Return asserted identity fields using runtime-facing types."""
+        subject = request_context.headers.get(self._header_subject, "")
+        email = request_context.headers.get(self._header_email)
+        groups_raw = request_context.headers.get(self._header_groups, "")
+        return subject, email, groups_raw
+
+    def _identity_payload_parts(self, request_context: RequestContext) -> tuple[str, str, str]:
+        """Return the asserted identity fields bound into the proxy signature."""
+        subject, email, groups_raw = self._identity_headers(request_context)
+        groups = ",".join(g.strip() for g in groups_raw.split(",") if g.strip())
+        return subject, email or "", groups
 
     def _is_from_trusted_proxy(self, request_context: RequestContext) -> bool:
         """Check if request came from a trusted proxy using CIDR matching."""
@@ -257,6 +274,42 @@ def _is_from_trusted_proxy(self, request_context: RequestContext) -> bool:
             pass
         return False
 
+    def _verify_proxy_signature(self, request_context: RequestContext) -> bool:
+        """Verify that the request was signed by a trusted proxy with the shared secret."""
+        if self._shared_secret is None:
+            return True
+        signature = request_context.headers.get(self._signature_header)
+        timestamp_str = request_context.headers.get(self._timestamp_header)
+        if not signature or not timestamp_str:
+            logger.debug("missing_proxy_signature", remote_addr=request_context.remote_addr)
+            return False
+        try:
+            timestamp = int(timestamp_str)
+        except ValueError:
+            logger.debug("invalid_proxy_timestamp", timestamp=timestamp_str)
+            return False
+        now = datetime.now(UTC).timestamp()
+        if abs(now - timestamp) > 300:
+            logger.debug("expired_proxy_timestamp", timestamp=timestamp_str)
+            return False
+        subject, email, groups = self._identity_payload_parts(request_context)
+        remote_addr = request_context.remote_addr or ""
+        path = request_context.path or ""
+        payload_parts: tuple[str, str, str, str, str, str] = (
+            str(timestamp),
+            remote_addr,
+            path,
+            subject,
+            email,
+            groups,
+        )
+        payload = ":".join(payload_parts)
+        expected = hmac.new(self._shared_secret.encode(), payload.encode(), "sha256").hexdigest()
+        if not hmac.compare_digest(signature, expected):
+            logger.debug("invalid_proxy_signature", remote_addr=request_context.remote_addr)
+            return False
+        return True
+
     def authenticate(self, request_context: RequestContext) -> AuthResult:
         """Authenticate using proxy-asserted identity."""
         if not self._is_from_trusted_proxy(request_context):
@@ -275,7 +328,23 @@ def authenticate(self, request_context: RequestContext) -> AuthResult:
                 reason_code="invalid_identity_payload",
             )
 
-        subject = request_context.headers.get(self._header_subject)
+        if not self._verify_proxy_signature(request_context):
+            _log_auth_event(
+                "failure",
+                None,
+                "invalid_identity_payload",
+                "proxy",
+                auth_method="proxy",
+                path=request_context.path,
+                remote_addr=request_context.remote_addr,
+                reason="invalid_signature",
+            )
+            return AuthResult(
+                authenticated=False,
+                reason_code="invalid_identity_payload",
+            )
+
+        subject, email, groups_raw = self._identity_headers(request_context)
         if not subject:
             _log_auth_event(
                 "failure",
@@ -291,8 +360,6 @@ def authenticate(self, request_context: RequestContext) -> AuthResult:
                 reason_code="missing_credentials",
             )
 
-        email = request_context.headers.get(self._header_email)
-        groups_raw = request_context.headers.get(self._header_groups, "")
         groups = tuple(g.strip() for g in groups_raw.split(",") if g.strip())
 
         principal = AuthPrincipal(
@@ -445,6 +512,18 @@ def _load_proxy_config() -> dict[str, Any]:
     if header_subject:
         config["header_subject"] = header_subject
 
+    header_email = os.environ.get("PHLO_AUTH_PROXY_HEADER_EMAIL")
+    if header_email:
+        config["header_email"] = header_email
+
+    header_groups = os.environ.get("PHLO_AUTH_PROXY_HEADER_GROUPS")
+    if header_groups:
+        config["header_groups"] = header_groups
+
+    shared_secret = os.environ.get("PHLO_AUTH_PROXY_SHARED_SECRET")
+    if shared_secret:
+        config["shared_secret"] = shared_secret
+
     return config
 
 

src/phlo/cli/commands/schema_migrate.py

@@ -61,7 +61,7 @@ def _resolve_migrator() -> Any:
                 f"[red]Configured schema migrator '{configured_migrator}' is not registered.[/red]"
             )
             console.print(f"Available schema migrators: {available}")
-            sys.exit(1)
+            raise SystemExit(1)
         return selected.provider
 
     default_table_store = configured_capability_name("table_store")

src/phlo/hooks/bus.py

@@ -193,10 +193,7 @@ def _matches_filters(filters: HookFilter, event: HookEvent) -> bool:
             event_asset_keys = _event_asset_keys(event)
             if not event_asset_keys or not filters.asset_keys.intersection(event_asset_keys):
                 return False
-        return not (
-            filters.tags is not None
-            and not all(event.tags.get(k) == v for k, v in filters.tags.items())
-        )
+        return filters.tags is None or all(event.tags.get(k) == v for k, v in filters.tags.items())
 
 
 def _event_asset_keys(event: HookEvent) -> set[str]:

src/phlo/infrastructure/config.py

@@ -25,15 +25,15 @@ def _default_project_root() -> Path:
     """Resolve the default project root from environment or current working directory."""
     project_root = os.environ.get("PHLO_PROJECT_PATH")
     if project_root:
-        resolved = Path(project_root).resolve()
-        if ".." in Path(project_root).parts:
+        raw_path = Path(project_root)
+        if ".." in raw_path.parts:
             logger.warning(
                 "project_path_traversal_rejected",
                 raw=project_root,
-                resolved=str(resolved),
+                reason="path_traversal_sequence",
             )
             raise ValueError(f"PHLO_PROJECT_PATH contains path traversal: {project_root}")
-        return resolved
+        return raw_path.resolve()
     return Path.cwd()