RBAC and Authorization

                           +---------------------+
                           |       ADMIN         |
                           |   (ADMIN_ALL        |
                           |    meta-permission) |
                           +----------+----------+
                                      |
                           +----------v----------+
                           |      REVIEWER       |
                           |  Code review +      |
                           |  GitHub + GitLab    |
                           +----------+----------+
                                      |
                           +----------v----------+
                           |      ANALYST        |
                           |  Query execution    |
                           |  + API keys         |
                           +----------+----------+
                                      |
                           +----------v----------+
                           |       VIEWER        |
                           |    Read-only        |
                           +---------------------+

from src.api.auth.permissions import Role, ROLE_PERMISSIONS, Permission

# ADMIN stores only the meta-permission
ROLE_PERMISSIONS[Role.ADMIN]  # => {Permission.ADMIN_ALL}

# get_role_permissions() expands ADMIN_ALL to all 21 permissions
from src.api.auth.permissions import get_role_permissions
perms = get_role_permissions(Role.ADMIN)  # => all Permission values

class TokenPayload(BaseModel):
    sub: str                          # Subject (user_id)
    jti: str                          # JWT ID (unique identifier)
    exp: datetime                     # Expiration time
    iat: datetime                     # Issued at
    type: str                         # "access" or "refresh"
    scopes: list[str] = []            # Permission scopes
    role: Optional[str] = None        # User role
    group_id: Optional[str] = None    # Optional group scope (for CI/CD service accounts)

from src.api.auth.jwt_handler import (
    create_access_token,   # Create JWT access token
    create_refresh_token,  # Create JWT refresh token
    decode_token,          # Decode and validate JWT
    verify_token,          # Verify JWT and check type
    get_token_jti,         # Extract JTI without full verification
    blacklist_token,       # Add token to blacklist
    is_token_blacklisted,  # Check if token is blacklisted
    load_blacklist_cache,  # Load blacklist from DB into memory
)

def create_access_token(
    user_id: str,
    scopes: Optional[list[str]] = None,
    role: Optional[str] = None,
    expires_delta: Optional[timedelta] = None,
    group_id: Optional[str] = None,
) -> str:

def get_token_jti(token: str) -> str:

# Get token
curl -X POST /api/v1/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username": "analyst", "password": "***"}'

# Response:
{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "refresh_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "bearer",
  "expires_in": 1800
}

# Use token
curl -X GET /api/v1/scenarios \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

# Logout - add token to blacklist
curl -X POST /api/v1/auth/logout \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

Prefix:  rag_<8 hex chars>     (e.g., rag_a1b2c3d4)
Secret:  <48 hex chars>        (secrets.token_hex(24))
Full:    rag_<8hex>_<48hex>

Example: rag_a1b2c3d4_e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8

def generate_api_key() -> tuple[str, str, str]:
    """Returns (full_key, prefix, key_hash)."""
    prefix = f"rag_{secrets.token_hex(4)}"   # e.g., rag_a1b2c3d4
    secret = secrets.token_hex(24)            # 48 hex chars
    full_key = f"{prefix}_{secret}"
    key_hash = hash_api_key(full_key)         # SHA-256
    return full_key, prefix, key_hash

def get_default_scopes_for_api_key() -> List[str]:
    return [
        "scenarios:read",
        "scenarios:execute",
        "query:execute",
        "sessions:read",
        "sessions:write",
        "history:read",
    ]

# Create API key
curl -X POST /api/v1/auth/api-keys \
  -H "Authorization: Bearer <admin_token>" \
  -d '{"name": "CI Pipeline", "scopes": ["query:execute", "scenarios:execute"]}'

# Response:
{
  "id": "key_123",
  "key": "rag_a1b2c3d4_e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8",
  "name": "CI Pipeline",
  "prefix": "rag_a1b2c3d4",
  "scopes": ["query:execute", "scenarios:execute"],
  "created_at": "2026-03-07T10:00:00Z",
  "expires_at": "2026-06-07T10:00:00Z",
  "is_revoked": false
}

# Use API key
curl -X GET /api/v1/scenarios \
  -H "X-API-Key: rag_a1b2c3d4_e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8"

codegraph auth service-account create --owner admin --name ci-bot --template ci --interface api
codegraph auth service-account list
codegraph auth service-account inspect --id <service-account-id>
codegraph auth service-account rotate --id <service-account-id>
codegraph auth service-account revoke --id <credential-id>
codegraph auth service-account deactivate --id <service-account-id>

oauth2:
  providers:
    keycloak:
      enabled: true
      client_id: "codegraph"
      client_secret: "${KEYCLOAK_SECRET}"
      authorize_url: "https://keycloak.company.com/realms/main/protocol/openid-connect/auth"
      token_url: "https://keycloak.company.com/realms/main/protocol/openid-connect/token"
      userinfo_url: "https://keycloak.company.com/realms/main/protocol/openid-connect/userinfo"
      scopes: ["openid", "profile", "email", "groups"]
      role_claim: "realm_access.roles"
      role_mapping:
        codegraph-admin: admin
        codegraph-reviewer: reviewer
        codegraph-analyst: analyst
        codegraph-viewer: viewer

ldap:
  enabled: true
  server: "ldaps://ldap.company.com:636"
  base_dn: "dc=company,dc=com"
  bind_dn: "cn=codegraph,ou=services,dc=company,dc=com"
  bind_password: "${LDAP_PASSWORD}"
  user_search_base: "ou=users,dc=company,dc=com"
  user_search_filter: "(sAMAccountName={username})"
  group_search_base: "ou=groups,dc=company,dc=com"
  group_membership_attr: "memberOf"
  group_mapping:
    "CN=CodeGraph-Admins,OU=Groups,DC=company,DC=com": admin
    "CN=CodeGraph-Reviewers,OU=Groups,DC=company,DC=com": reviewer
    "CN=CodeGraph-Analysts,OU=Groups,DC=company,DC=com": analyst
    "CN=CodeGraph-Viewers,OU=Groups,DC=company,DC=com": viewer
  sync_interval_minutes: 15

# IAM authentication flow in get_auth_context():
iam_token = request.headers.get("X-YC-IAM-Token")
if iam_token:
    validator = get_iam_validator()
    user_info = await validator.validate_token(iam_token)
    # Returns AuthContext with:
    #   role=Role.ANALYST
    #   scopes=["scenarios:read", "query:execute", "review:execute"]
    #   auth_method="iam"

# Authenticate with IAM token
curl -X GET /api/v1/scenarios \
  -H "X-YC-IAM-Token: t1.9euelZqMkJKVjJqTnZyRm5GUkZqRke..."

from src.api.auth.middleware import (
    get_auth_context,        # Get context (returns unauthenticated context if no credentials)
    get_current_user,        # FastAPI dependency, raises 401 if unauthenticated
    get_optional_user,       # FastAPI dependency, returns unauthenticated context if no credentials
    require_auth,            # Require any authentication (raises 401)
    require_permission,      # Factory: require specific permission (raises 403)
    require_any_permission,  # Factory: require any of multiple permissions (raises 403)
    require_role,            # Factory: require specific role (raises 403)
    require_admin,           # Shortcut for require_role(Role.ADMIN)
    require_analyst,         # Shortcut for require_role(Role.ANALYST, Role.REVIEWER, Role.ADMIN)
    require_reviewer,        # Shortcut for require_role(Role.REVIEWER, Role.ADMIN)
)

async def get_auth_context(request, bearer_token, api_key) -> AuthContext
async def get_current_user(auth: AuthContext = Depends(require_auth)) -> AuthContext
async def get_optional_user(auth: AuthContext = Depends(get_auth_context)) -> AuthContext
async def require_auth(auth: AuthContext = Depends(get_auth_context)) -> AuthContext
def require_permission(permission: Permission) -> Callable  # dependency factory
def require_any_permission(*permissions: Permission) -> Callable  # dependency factory
def require_role(*roles: Role) -> Callable  # dependency factory

from fastapi import Depends, APIRouter
from src.api.auth.middleware import (
    require_permission,
    require_any_permission,
    require_admin,
    get_current_user,
    get_optional_user,
)
from src.api.auth.permissions import Permission

router = APIRouter()

# Require specific permission
@router.post("/query/execute")
async def execute_query(
    query: str,
    auth = Depends(require_permission(Permission.QUERY_EXECUTE))
):
    # auth.user_id, auth.role, auth.scopes, auth.group_id available
    return {"result": "..."}

# Require one of multiple permissions
@router.get("/reviews")
async def list_reviews(
    auth = Depends(require_any_permission(
        Permission.REVIEW_EXECUTE,
        Permission.REVIEW_GITHUB,
        Permission.REVIEW_GITLAB
    ))
):
    return {"reviews": [...]}

# Require Admin role
@router.delete("/users/{user_id}")
async def delete_user(
    user_id: str,
    auth = Depends(require_admin)
):
    return {"deleted": user_id}

# Get authenticated user (raises 401 if not authenticated)
@router.get("/me")
async def get_profile(auth = Depends(get_current_user)):
    return {"user_id": auth.user_id, "role": auth.role}

# Optional authentication (no error if unauthenticated)
@router.get("/public")
async def public_endpoint(auth = Depends(get_optional_user)):
    if auth.is_authenticated:
        return {"greeting": f"Hello, {auth.username}"}
    return {"greeting": "Hello, anonymous"}

class AuthContext:
    def __init__(
        self,
        user_id: Optional[str] = None,
        username: Optional[str] = None,
        role: Optional[Role] = None,
        scopes: Optional[List[str]] = None,
        auth_method: str = "none",       # "jwt", "api_key", "oauth2", "ldap", "iam", "none"
        group_id: Optional[str] = None,  # optional group scope (from JWT or API key)
    ):
        ...

    @property
    def is_authenticated(self) -> bool:
        """True if user_id is not None."""

    def has_permission(self, permission: Permission) -> bool:
        """Check if user has a specific permission."""

from src.api.auth.permissions import (
    get_role_permissions,
    has_permission,
    has_any_permission,
    has_all_permissions,
    validate_scopes,
    get_default_scopes_for_api_key,
)

def has_permission(
    role: Optional[Role],
    required_permission: Permission,
    user_scopes: Optional[List[str]] = None,
) -> bool:

def has_any_permission(
    role: Optional[Role],
    required_permissions: List[Permission],
    user_scopes: Optional[List[str]] = None,
) -> bool:

def has_all_permissions(
    role: Optional[Role],
    required_permissions: List[Permission],
    user_scopes: Optional[List[str]] = None,
) -> bool:

def validate_scopes(scopes: List[str]) -> List[str]:
    valid_permissions = {p.value for p in Permission}
    return [s for s in scopes if s in valid_permissions]

# config.yaml
security:
  api_key_scope_enforcement: true   # Default: enabled

from src.api.auth.scope_enforcement import require_scope

@router.get("/admin/stats")
async def admin_stats(auth: AuthContext = Depends(require_scope("admin:read"))):
    ...

def require_scope(scope: str):
    """FastAPI dependency factory that checks if the authenticated caller has a scope.

    Raises:
        HTTPException(401) if not authenticated
        HTTPException(403) if scope is missing
    """

def check_api_key_scope(auth: AuthContext, request_path: str) -> Optional[str]:
    """Check if an API-key-authenticated request has the required scope.

    Returns None if scope is satisfied, or an error message string if missing.
    """

async def verify_webhook_signature(
    request: Request,
    secret: Optional[str] = None,
    platform: str = "sourcecraft",
    max_age_seconds: int = 300,
) -> bytes:

from src.api.auth.webhook_auth import verify_webhook_signature

@router.post("/webhooks/sourcecraft")
async def handle_sourcecraft_webhook(request: Request):
    body = await verify_webhook_signature(
        request,
        secret="your-webhook-secret",
        platform="sourcecraft",
        max_age_seconds=300,
    )
    payload = json.loads(body)
    ...

async def blacklist_token(jti: str, expires_at: Optional[datetime] = None) -> None:

async def is_token_blacklisted(jti: str) -> bool:

async def load_blacklist_cache() -> int:

async def _blacklist_sync_task(interval_seconds: int = 60) -> None:

Request                              blacklist_token()
   |                                       |
   v                                       v
is_token_blacklisted()              _blacklisted_tokens (set)
   |                                  +    |
   |  1. check in-memory set          |    v
   |  2. fallback to PostgreSQL        | PostgreSQL: TokenBlacklist table
   |     (adds to cache if found)      |
   v                                   |
 True/False                     _blacklist_sync_task()
                                (periodic refresh)

class ApiKeyInfo(BaseModel):
    id: str                            # Unique key identifier
    name: str                          # Human-readable name
    prefix: str                        # Key prefix (e.g., "rag_a1b2c3d4")
    scopes: List[str]                  # Permission scopes
    created_at: datetime               # Creation timestamp
    expires_at: Optional[datetime]     # Expiration (None = never)
    last_used_at: Optional[datetime]   # Last usage timestamp
    is_revoked: bool                   # Revocation status

class ApiKeyWithSecret(ApiKeyInfo):
    key: str    # Full API key (e.g., "rag_a1b2c3d4_<48hex>")

from src.api.auth.api_keys import (
    generate_api_key,      # -> (full_key, prefix, key_hash)
    hash_api_key,          # key -> SHA-256 hex digest
    verify_api_key,        # (key, stored_hash) -> bool (constant-time)
    extract_prefix,        # key -> prefix string
    is_key_expired,        # expires_at -> bool
    calculate_expiration,  # days -> Optional[datetime]
    validate_api_key,      # Full validation pipeline (hash + revoked + expired)
)

{
  "timestamp": "2026-03-07T10:30:00.000Z",
  "event_type": "AUTH_SUCCESS",
  "user_id": "user_123",
  "username": "analyst@company.com",
  "role": "analyst",
  "auth_method": "jwt",
  "group_id": null,
  "ip_address": "10.0.0.50",
  "user_agent": "Mozilla/5.0...",
  "request_path": "/api/v1/scenarios",
  "request_method": "GET"
}

from src.api.auth.project_auth import require_project_access

# Require at least editor role within the project group
@router.post("/projects/{project_id}/import")
async def import_project(
    project_id: str,
    auth = Depends(require_project_access("editor"))
):
    ...

# config.yaml
multi_tenant:
  enabled: false    # Default: disabled

python scripts/migrate_default_group.py          # Creates "default" group, assigns all users as ADMIN
python scripts/migrate_default_group.py --dry-run # Preview changes without applying

# config.yaml
security:
  path_validation_enabled: true       # Default: enabled
  path_validation_deny_symlinks: true  # Reject symlinks outside whitelist
  path_validation_deny_relative: true  # Reject relative paths
  path_validation_allowed_base_dirs:
    - "data/projects/"