File size: 8,160 Bytes
db4ba8d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
7ffa183
db4ba8d
 
 
 
 
dd9584b
db4ba8d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
6a25d7a
db4ba8d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
6a25d7a
db4ba8d
 
 
 
 
 
e216e82
 
b721542
e216e82
 
 
 
b721542
6a25d7a
e216e82
 
db4ba8d
 
 
 
 
 
6a25d7a
 
 
 
 
 
 
 
 
db4ba8d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
"""
TradeFlow AI — FastAPI Dependencies

Auth (Keycloak JWT), DB session, tier/role guards.
PRD §4 Decision 2: Keycloak 26 is the ONLY auth provider.
"""

from __future__ import annotations

import time
from typing import Annotated

import httpx
import structlog
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from jose.exceptions import ExpiredSignatureError

try:
    from supabase import AsyncClient, acreate_client
except Exception:  # pragma: no cover - optional in lightweight test environments
    AsyncClient = None
    acreate_client = None

from .config import settings

log = structlog.get_logger()

# ── Supabase client (singleton) ───────────────────────────────────────────────
_supabase_client: AsyncClient | None = None


async def init_supabase() -> None:
    global _supabase_client
    if acreate_client is None:
        log.info("Supabase client not available in this environment; skipping initialization")
        _supabase_client = None
        return

    _supabase_client = await acreate_client(
        settings.SUPABASE_URL,
        settings.SUPABASE_SERVICE_KEY.get_secret_value(),  # Service key for server-side ops
    )
    log.info("Supabase client initialized")


async def close_supabase() -> None:
    global _supabase_client
    if _supabase_client:
        pass
        _supabase_client = None


def get_supabase() -> AsyncClient:
    if _supabase_client is None:
        raise RuntimeError("Supabase client not initialized. Call init_supabase() first.")
    return _supabase_client


# ── Keycloak JWKS cache ───────────────────────────────────────────────────────
_keycloak_jwks: dict | None = None
_keycloak_jwks_time: float = 0
KEYCLOAK_JWKS_TTL = 3600  # Refresh every hour


def get_keycloak_jwks() -> dict:
    """Fetch Keycloak JWKS with TTL-based caching (refresh every hour)."""
    global _keycloak_jwks, _keycloak_jwks_time
    now = time.time()

    if not _keycloak_jwks or (now - _keycloak_jwks_time) > KEYCLOAK_JWKS_TTL:
        with httpx.Client() as client:
            response = client.get(settings.KEYCLOAK_JWKS_URL)
            response.raise_for_status()
            _keycloak_jwks = response.json()
            _keycloak_jwks_time = now
            log.info("Refreshed Keycloak JWKS cache")

    return _keycloak_jwks


# ── JWT Bearer scheme ─────────────────────────────────────────────────────────
bearer_scheme = HTTPBearer(auto_error=False)


class CurrentUser:
    """Decoded Keycloak JWT claims, enriched with Supabase profile."""

    def __init__(
        self,
        sub: str,
        email: str,
        full_name: str,
        roles: list[str],
        tier: str,
        company_id: str | None,
        raw_token: str,
    ) -> None:
        self.id = sub
        self.sub = sub
        self.email = email
        self.full_name = full_name
        self.roles = roles
        self.tier = tier
        self.company_id = company_id
        self.raw_token = raw_token

    @property
    def is_enterprise(self) -> bool:
        return self.tier == "enterprise"

    @property
    def is_admin(self) -> bool:
        return "admin" in self.roles or "supervisor" in self.roles

    @property
    def role(self) -> str:
        """Primary role (first in list)."""
        return self.roles[0] if self.roles else "operator"


async def get_current_user(
    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)],
    supabase: Annotated[AsyncClient, Depends(get_supabase)],
) -> CurrentUser:
    """
    Validate Keycloak RS256 JWT and return enriched user.
    PRD §4 Decision 2: verify against Keycloak JWKS endpoint.
    """
    if settings.DISABLE_AUTH:
        return CurrentUser(
            sub=None,  # type: ignore
            email="admin@tradeflow.ai",
            full_name="Demo User",
            roles=["admin", "operator", "supervisor", "sme"],
            tier="enterprise",
            company_id=None,
            raw_token="demo-token",
        )

    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid or expired authentication token",
        headers={"WWW-Authenticate": "Bearer"},
    )

    if not credentials:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )

    token = credentials.credentials

    try:
        jwks = get_keycloak_jwks()
        payload = jwt.decode(
            token,
            jwks,
            algorithms=["RS256"],
            audience=settings.KEYCLOAK_CLIENT_ID,
            issuer=settings.KEYCLOAK_ISSUER,
        )
    except ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has expired",
            headers={"WWW-Authenticate": "Bearer"},
        )
    except JWTError as e:
        log.warning("JWT validation failed", error=str(e))
        raise credentials_exception

    sub: str = payload.get("sub", "")
    if not sub:
        raise credentials_exception

    # Extract Keycloak realm roles
    realm_access = payload.get("realm_access", {})
    roles: list[str] = realm_access.get("roles", [])
    # Filter to only TradeFlow roles
    tradeflow_roles = [r for r in roles if r in ("operator", "admin", "supervisor", "importer")]

    # Fetch profile from Supabase for tier + company_id
    try:
        result = await supabase.table("profiles").select(
            "id, full_name, email, tier, role, company_id"
        ).eq("id", sub).single().execute()
        profile = result.data
        tier = profile.get("tier", "sme")
        company_id = profile.get("company_id")
        full_name = profile.get("full_name", payload.get("name", ""))
        email = profile.get("email", payload.get("email", ""))
    except Exception:
        # Profile not yet created — use JWT claims as fallback
        tier = "sme"
        company_id = sub  # Fallback to user's own ID so they can act as their own company
        full_name = payload.get("name", "")
        email = payload.get("email", "")

    return CurrentUser(
        sub=sub,
        email=email,
        full_name=full_name,
        roles=tradeflow_roles if tradeflow_roles else ["operator"],
        tier=tier,
        company_id=company_id,
        raw_token=token,
    )


# ── Role/tier guards ──────────────────────────────────────────────────────────

async def require_enterprise(
    user: Annotated[CurrentUser, Depends(get_current_user)],
) -> CurrentUser:
    """Guard: Enterprise tier only."""
    if not user.is_enterprise:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="This feature requires an Enterprise tier subscription.",
        )
    return user


async def require_admin(
    user: Annotated[CurrentUser, Depends(get_current_user)],
) -> CurrentUser:
    """Guard: Admin or Supervisor role only."""
    if not user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="This action requires Administrator or Supervisor role.",
        )
    return user


async def require_operator(
    user: Annotated[CurrentUser, Depends(get_current_user)],
) -> CurrentUser:
    """Guard: Any authenticated user with operator/admin/supervisor role."""
    if "importer" in user.roles and len(user.roles) == 1:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Importers cannot perform operator actions.",
        )
    return user