Authentication Architecture

# federation_router.py
state_payload = {
    "tenant_id": str(tenant_obj.id),
    "tenant_slug": tenant_obj.slug or tenant_obj.name,
    "frontend_state": state or "",
    "nonce": secrets.token_hex(16),        # 32 hex chars
    "redirect_uri": redirect_uri,
    "correlation_id": correlation_id,      # 16 hex chars
    "exp": int(time.time()) + 600,         # 10 minute TTL
    "iat": int(time.time()),
    "config_version": tenant_config_version,
}
signed_state = pyjwt.encode(state_payload, settings.jwt_secret, algorithm="HS256")

# Cache key format
state_cache_key = f"oidc:state:{nonce}"
 
# Cached payload (5 fields)
state_cache_payload = {
    "tenant_id": str(tenant_obj.id),
    "tenant_slug": tenant_obj.slug or tenant_obj.name,
    "redirect_uri": redirect_uri,
    "config_version": tenant_config_version,
    "iat": state_payload["iat"],
}
 
# Storage with TTL
await redis_client.setex(
    state_cache_key,
    settings.oidc_state_ttl_seconds,  # 600 seconds
    json.dumps(state_cache_payload)
)

# federation_router.py
grace_period = min(
    settings.oidc_redirect_grace_period_seconds,  # 900 seconds (15 min)
    settings.oidc_state_ttl_seconds,               # 600 seconds (10 min)
)

# federation_router.py
def _select_auth_method(methods: list[str] | None) -> str | None:
    if not methods:
        return None
    normalized = [str(m).lower() for m in methods if m]
    if "client_secret_post" in normalized:
        return "client_secret_post"
    if "client_secret_basic" in normalized:
        return "client_secret_basic"
    return normalized[0]

# federation_router.py
if tenant_obj.state != TenantState.ACTIVE:
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="Authentication unavailable - tenant is not active",
    )

# federation_router.py
# Check tenant_id matches
if expected_tenant_id and expected_tenant_id != str(tenant_id):
    # Event: callback.state_tampered, reason: tenant_id_mismatch
    raise HTTPException(403, "Invalid state - security validation failed")
 
# Check tenant_slug matches (case-insensitive)
if expected_tenant_slug and expected_tenant_slug != tenant_slug.lower():
    # Event: callback.state_tampered, reason: tenant_slug_mismatch
    raise HTTPException(403, "Invalid state - security validation failed")

-- tenant_table.py
ALTER TABLE tenants ADD COLUMN federation_config JSONB NOT NULL DEFAULT '{}';

federation_config = {
    "provider": "entra_id",                    # Identity provider label
    "client_id": "...",                        # OAuth client ID
    "client_secret": "enc:fernet:v1:...",      # Encrypted with Fernet
    "discovery_endpoint": "https://...",       # OIDC discovery URL
    "issuer": "...",                           # From discovery
    "authorization_endpoint": "...",           # From discovery
    "token_endpoint": "...",                   # From discovery
    "jwks_uri": "...",                         # From discovery
    "scopes": ["openid", "email", "profile"],  # Default scopes
    "allowed_domains": ["example.com"],        # Email domain whitelist
    "canonical_public_origin": "https://...",  # Tenant's access URL
    "redirect_path": "/login/callback",        # Callback path
    "encrypted_at": "2025-01-07T12:00:00Z",    # When secret was encrypted
}

uv run python -m intric.cli.generate_encryption_key

claims_mapping = federation_config.get("claims_mapping", {"email": "email"})
email_claim = claims_mapping.get("email", "email")
email = payload.get(email_claim)

email = id_token.get("email")
domain = email.split("@")[1].lower()  # Case-insensitive
if domain not in [d.lower() for d in allowed_domains]:
    raise HTTPException(403, "Domain not allowed")

# federation_router.py
jwk_client = PyJWKClient(jwks_uri)
signing_key = jwk_client.get_signing_key_from_jwt(id_token).key