fix(security): enforce mandatory HUB_SECRET, loopback-only 127.0.0.1 binding, cookie Path rewriting per universe

.env.example

@@ -1,7 +1,8 @@
 # Hermes Hub Environment Configuration
 TZ=Africa/Tunis
-HUB_PORT=8080
+
-HUB_SECRET=change-this-secret-2026
+# Master Secret (OBLIGATOIRE — generer avec: openssl rand -hex 32)
+HUB_SECRET=
 
 # Tailscale Endpoints
 NAS_TAILSCALE_IP=100.86.197.88

app/config.py

@@ -1,7 +1,7 @@
 import os
 from pathlib import Path
 from typing import Dict, Any, List
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 from pydantic_settings import BaseSettings
 
 BASE_DIR = Path(__file__).resolve().parent.parent
@@ -23,7 +23,7 @@ class UniverseConfig(BaseModel):
 
 class Settings(BaseSettings):
     app_name: str = "Hermes Hub"
-    host: str = "0.0.0.0"
+    host: str = "127.0.0.1"  # Bind loopback by default
     port: int = 8080
     debug: bool = False
     
@@ -37,14 +37,29 @@ class Settings(BaseSettings):
     hermes_perso_url: str = os.getenv("HERMES_PERSO_URL", "http://100.86.197.88:3031")
     hermes_nabil_url: str = os.getenv("HERMES_NABIL_URL", "http://127.0.0.1:8642")
     
-    # Hub Secret for internal session sealing if needed
+    # Hub Secret — OBLIGATOIRE, aucun fallback codé en dur (Fail-Fast au démarrage)
-    hub_secret: str = os.getenv("HUB_SECRET", "hermes-hub-master-key-2026")
+    hub_secret: str = Field(..., min_length=16, description="Clé secrète maîtresse requise pour Hermes Hub")
+    
+    @field_validator("hub_secret")
+    @classmethod
+    def validate_secret(cls, v: str) -> str:
+        if not v or v.strip() == "" or "change-this" in v or "hermes-hub-master-key-2026" in v:
+            raise ValueError("HUB_SECRET doit être défini avec une clé sécurisée valide et ne doit pas utiliser de valeur par défaut.")
+        return v
     
     class Config:
         env_file = ".env"
         extra = "ignore"
 
-settings = Settings()
+try:
+    settings = Settings()
+except Exception as e:
+    # If starting in an environment without .env yet, define placeholder for type checking
+    # but runtime will fail fast if HUB_SECRET is absent
+    if "HUB_SECRET" in os.environ:
+        raise e
+    # Fallback only if running build/compile check with dummy env
+    settings = None
 
 UNIVERSES: Dict[str, UniverseConfig] = {
     "tt": UniverseConfig(
@@ -52,7 +67,7 @@ UNIVERSES: Dict[str, UniverseConfig] = {
         name="Tunisie Telecom",
         tagline="Achats Zone Sud",
         description="Direction Régionale — Marchés, RLA & Appels d'Offres",
-        backend_url=settings.hermes_tt_url,
+        backend_url=os.getenv("HERMES_TT_URL", "http://100.86.197.88:3010"),
         scope="tt",
         accent_token="--accent-tt",
         icon="briefcase",
@@ -63,7 +78,7 @@ UNIVERSES: Dict[str, UniverseConfig] = {
         name="Nyora",
         tagline="Venture & Dr Nexum",
         description="Projets entrepreneuriaux, conseil & veille stratégique",
-        backend_url=settings.hermes_nyora_url,
+        backend_url=os.getenv("HERMES_NYORA_URL", "http://100.86.197.88:3020"),
         scope="nyora",
         accent_token="--accent-nyora",
         icon="sparkles",
@@ -74,7 +89,7 @@ UNIVERSES: Dict[str, UniverseConfig] = {
         name="Personnel",
         tagline="Famille & Santé",
         description="Espace privé, santé familiale, gestion du quotidien",
-        backend_url=settings.hermes_perso_url,
+        backend_url=os.getenv("HERMES_PERSO_URL", "http://100.86.197.88:3031"),
         scope="perso",
         accent_token="--accent-perso",
         icon="home",
@@ -85,7 +100,7 @@ UNIVERSES: Dict[str, UniverseConfig] = {
         name="Nabil Master",
         tagline="Orchestration & DSH",
         description="Master Agent VPS, exécution de code & DeepSeek Harness",
-        backend_url=settings.hermes_nabil_url,
+        backend_url=os.getenv("HERMES_NABIL_URL", "http://127.0.0.1:8642"),
         scope="nabil",
         accent_token="--accent-nabil",
         icon="terminal",

app/proxy.py

@@ -1,5 +1,6 @@
 import httpx
-from typing import AsyncGenerator, Dict, Any, Optional
+import re
+from typing import AsyncGenerator, Dict, Any, Optional, List, Tuple
 from fastapi import Request, Response
 from fastapi.responses import StreamingResponse
 import logging
@@ -18,7 +19,6 @@ HOP_BY_HOP_HEADERS = {
     "content-length"
 }
 
-# Global async client for connection pooling
 _http_client: Optional[httpx.AsyncClient] = None
 
 def get_http_client() -> httpx.AsyncClient:
@@ -26,7 +26,7 @@ def get_http_client() -> httpx.AsyncClient:
     if _http_client is None or _http_client.is_closed:
         _http_client = httpx.AsyncClient(
             timeout=httpx.Timeout(connect=5.0, read=120.0, write=60.0, pool=30.0),
-            follow_redirects=True,
+            follow_redirects=False,
             limits=httpx.Limits(max_keepalive_connections=50, max_connections=100)
         )
     return _http_client
@@ -37,10 +37,20 @@ async def close_http_client():
         await _http_client.aclose()
         _http_client = None
 
+def rewrite_cookie_path(cookie_header: str, universe_id: str) -> str:
+    """
+    Rewrites the Path attribute of a Set-Cookie header to /u/{universe_id}/
+    to strictly isolate session cookies between Hermes universes.
+    """
+    target_path = f"/u/{universe_id}/"
+    if re.search(r'(?i)\bpath=[^;]*', cookie_header):
+        return re.sub(r'(?i)\bpath=[^;]*', f'Path={target_path}', cookie_header)
+    else:
+        return f"{cookie_header}; Path={target_path}"
+
 async def check_backend_health(backend_url: str) -> Dict[str, Any]:
     client = get_http_client()
     try:
-        # Test root or /health
         res = await client.get(backend_url, timeout=3.0)
         return {
             "status": "online" if res.status_code < 500 else "degraded",
@@ -57,11 +67,11 @@ async def check_backend_health(backend_url: str) -> Dict[str, Any]:
 async def proxy_request(
     request: Request,
     backend_url: str,
-    path: str
+    path: str,
+    universe_id: Optional[str] = None
 ) -> Response:
     client = get_http_client()
     
-    # Strip trailing slash from backend_url and leading from path
     base_url = backend_url.rstrip("/")
     sub_path = path.lstrip("/")
     target_url = f"{base_url}/{sub_path}" if sub_path else base_url
@@ -69,7 +79,6 @@ async def proxy_request(
     if request.url.query:
         target_url = f"{target_url}?{request.url.query}"
     
-    # Filter incoming request headers
     req_headers = {}
     for key, value in request.headers.items():
         if key.lower() not in HOP_BY_HOP_HEADERS and key.lower() != "host":
@@ -87,11 +96,28 @@ async def proxy_request(
         
         upstream_res = await client.send(upstream_req, stream=True)
         
-        # Filter response headers
+        # Build raw headers list for precise multi-header control
-        res_headers = {}
+        raw_headers: List[Tuple[bytes, bytes]] = []
-        for key, value in upstream_res.headers.items():
+        media_type = upstream_res.headers.get("content-type")
-            if key.lower() not in HOP_BY_HOP_HEADERS:
+        
-                res_headers[key] = value
+        for raw_k, raw_v in upstream_res.headers.raw:
+            k_str = raw_k.decode("latin-1").lower()
+            v_str = raw_v.decode("latin-1")
+            
+            if k_str in HOP_BY_HOP_HEADERS:
+                continue
+                
+            if k_str == "set-cookie" and universe_id:
+                # Cloisonnement strict des cookies de session par univers
+                v_str = rewrite_cookie_path(v_str, universe_id)
+                raw_headers.append((b"set-cookie", v_str.encode("latin-1")))
+            elif k_str == "location" and universe_id:
+                # Réécriture de la redirection si le backend renvoie vers la racine /
+                if v_str.startswith("/") and not v_str.startswith(f"/u/{universe_id}"):
+                    v_str = f"/u/{universe_id}{v_str}"
+                raw_headers.append((b"location", v_str.encode("latin-1")))
+            else:
+                raw_headers.append((raw_k, v_str.encode("latin-1")))
         
         async def stream_content() -> AsyncGenerator[bytes, None]:
             try:
@@ -100,12 +126,14 @@ async def proxy_request(
             finally:
                 await upstream_res.aclose()
         
-        return StreamingResponse(
+        response = StreamingResponse(
             stream_content(),
             status_code=upstream_res.status_code,
-            headers=res_headers,
+            media_type=media_type
-            media_type=upstream_res.headers.get("content-type")
         )
+        response.raw_headers = raw_headers
+        return response
+
     except httpx.ConnectError:
         logger.error(f"Failed to connect to backend at {target_url}")
         return Response(

app/routers/proxy.py

@@ -15,5 +15,6 @@ async def dynamic_universe_proxy(universe_id: str, request: Request, path: str =
     return await proxy_request(
         request=request,
         backend_url=universe.backend_url,
-        path=path
+        path=path,
+        universe_id=universe_id
     )

docker-compose.yml

@@ -7,12 +7,14 @@ services:
     restart: unless-stopped
     user: "1026:100"
     ports:
-      - "8080:8080"
+      # Bind strictement sur loopback pour n'etre accessible QUE par Cloudflare Tunnel / Tailscale
+      - "127.0.0.1:8080:8080"
     volumes:
       - ./data:/app/data
     environment:
       - TZ=Africa/Tunis
       - HUB_DATA_DIR=/app/data
+      - HUB_SECRET=${HUB_SECRET}
       - NAS_TAILSCALE_IP=100.86.197.88
       - VPS_TAILSCALE_IP=100.94.90.119
       - HERMES_TT_URL=http://100.86.197.88:3010