dragonked2/alphacode/auth

// Access Token (short-lived: 15 min)
{
  "sub": "user-123",
  "email": "user@example.com",
  "roles": ["user"],
  "type": "access",
  "iat": 1700000000,
  "exp": 1700000900
}
// Refresh Token (long-lived: 7 days)
{
  "sub": "user-123",
  "type": "refresh",
  "jti": "unique-token-id",
  "iat": 1700000000,
  "exp": 1700604800
}

# services/auth.py
from datetime import datetime, timedelta
from jose import jwt, JWTError
import secrets
class AuthService:
    def __init__(self, db, redis, config):
        self.db = db
        self.redis = redis
        self.config = config
    async def login(self, email: str, password: str) -> dict:
        user = await self.db.get_user_by_email(email)
        if not user or not verify_password(password, user.hashed_password):
            raise UnauthorizedError("Invalid credentials")
        access_token = self.create_access_token(user)
        refresh_token = await self.create_refresh_token(user)
        return {
            "access_token": access_token,
            "refresh_token": refresh_token,
            "token_type": "bearer",
            "expires_in": self.config.jwt_expires_in,
        }
    def create_access_token(self, user) -> str:
        payload = {
            "sub": str(user.id),
            "email": user.email,
            "roles": [user.role],
            "type": "access",
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + timedelta(minutes=15),
        }
        return jwt.encode(payload, self.config.jwt_secret, algorithm="HS256")
    async def create_refresh_token(self, user) -> str:
        jti = secrets.token_urlsafe(32)
        payload = {
            "sub": str(user.id),
            "type": "refresh",
            "jti": jti,
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + timedelta(days=7),
        }
        token = jwt.encode(payload, self.config.jwt_secret, algorithm="HS256")
        # Store in Redis for revocation
        await self.redis.setex(
            f"refresh_token:{jti}",
            timedelta(days=7),
            str(user.id),
        )
        return token
    async def refresh(self, refresh_token: str) -> dict:
        try:
            payload = jwt.decode(
                refresh_token, self.config.jwt_secret, algorithms=["HS256"]
            )
        except JWTError:
            raise UnauthorizedError("Invalid refresh token")
        if payload.get("type") != "refresh":
            raise UnauthorizedError("Invalid token type")
        jti = payload.get("jti")
        user_id = await self.redis.get(f"refresh_token:{jti}")
        if not user_id:
            raise UnauthorizedError("Refresh token revoked or expired")
        # Rotate: delete old, create new
        await self.redis.delete(f"refresh_token:{jti}")
        user = await self.db.get_user(int(user_id))
        new_access = self.create_access_token(user)
        new_refresh = await self.create_refresh_token(user)
        return {
            "access_token": new_access,
            "refresh_token": new_refresh,
            "token_type": "bearer",
        }
    async def logout(self, refresh_token: str):
        try:
            payload = jwt.decode(
                refresh_token, self.config.jwt_secret, algorithms=["HS256"]
            )
            jti = payload.get("jti")
            await self.redis.delete(f"refresh_token:{jti}")
        except JWTError:
            pass  # Token already invalid

# Middleware to check revoked tokens
async def check_token_revoked(token: str, redis) -> bool:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        jti = payload.get("jti")
        if jti:
            return await redis.exists(f"revoked_token:{jti}")
        return False
    except JWTError:
        return True

# services/oauth.py
import hashlib
import base64
import secrets
from urllib.parse import urlencode
class OAuthService:
    def __init__(self, config):
        self.config = config
    def get_authorization_url(self) -> dict:
        code_verifier = secrets.token_urlsafe(32)
        code_challenge = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode()).digest()
        ).rstrip(b"=").decode()
        params = {
            "client_id": self.config.client_id,
            "redirect_uri": self.config.redirect_uri,
            "response_type": "code",
            "scope": "openid profile email",
            "state": secrets.token_urlsafe(16),
            "code_challenge": code_challenge,
            "code_challenge_method": "S256",
        }
        return {
            "url": f"{self.config.auth_url}?{urlencode(params)}",
            "state": params["state"],
            "code_verifier": code_verifier,
        }
    async def exchange_code(self, code: str, code_verifier: str) -> dict:
        data = {
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": self.config.redirect_uri,
            "client_id": self.config.client_id,
            "client_secret": self.config.client_secret,
            "code_verifier": code_verifier,
        }
        response = httpx.post(self.config.token_url, data=data)
        return response.json()

async def get_client_token(client_id: str, client_secret: str) -> str:
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "api:read api:write",
    }
    response = httpx.post(token_url, data=data)
    token_data = response.json()
    return token_data["access_token"]

# handlers/auth.py
from fastapi import APIRouter, Request
from fastapi.responses import RedirectResponse
router = APIRouter()
@router.get("/auth/callback")
async def oauth_callback(request: Request, code: str, state: str):
    # Validate state
    stored_state = await request.session.get("oauth_state")
    if state != stored_state:
        raise HTTPException(400, "Invalid state parameter")
    # Exchange code for tokens
    token_data = await oauth_service.exchange_code(
        code,
        request.session.get("code_verifier"),
    )
    # Get or create user from ID token
    user_info = decode_id_token(token_data["id_token"])
    user = await get_or_create_user(user_info)
    # Create our own tokens
    access_token = auth_service.create_access_token(user)
    refresh_token = await auth_service.create_refresh_token(user)
    return RedirectResponse(
        url=f"{FRONTEND_URL}/auth/callback?token={access_token}",
        status_code=302,
    )

# middleware/session.py
from starlette.middleware.base import BaseHTTPMiddleware
import uuid
class SessionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        session_id = request.cookies.get("session_id")
        if session_id:
            session_data = await redis.get(f"session:{session_id}")
            if session_data:
                request.state.session = json.loads(session_data)
            else:
                session_id = None
        if not session_id:
            session_id = str(uuid.uuid4())
            request.state.session = {}
        response = await call_next(request)
        # Set session cookie
        response.set_cookie(
            "session_id",
            session_id,
            httponly=True,
            secure=True,
            samesite="lax",
            max_age=86400 * 7,  # 7 days
        )
        # Store session
        await redis.setex(
            f"session:{session_id}",
            86400 * 7,
            json.dumps(request.state.session),
        )
        return response

from itsdangerous import URLSafeTimedSerializer
class CookieSession:
    def __init__(self, secret_key: str):
        self.serializer = URLSafeTimedSerializer(secret_key)
    def create_session(self, user_id: int) -> str:
        data = {"user_id": user_id, "created_at": time.time()}
        return self.serializer.dumps(data)
    def verify_session(self, token: str, max_age: int = 604800) -> dict:
        try:
            return self.serializer.loads(token, max_age=max_age)
        except Exception:
            return None

# models/permissions.py
from enum import Enum
class Permission(str, Enum):
    USER_READ = "user:read"
    USER_WRITE = "user:write"
    USER_DELETE = "user:delete"
    ORDER_READ = "order:read"
    ORDER_WRITE = "order:write"
    REPORT_VIEW = "report:view"
    ADMIN_ALL = "admin:*"
ROLE_PERMISSIONS = {
    "viewer": [Permission.USER_READ, Permission.ORDER_READ],
    "user": [
        Permission.USER_READ,
        Permission.USER_WRITE,
        Permission.ORDER_READ,
        Permission.ORDER_WRITE,
    ],
    "admin": [
        Permission.USER_READ,
        Permission.USER_WRITE,
        Permission.USER_DELETE,
        Permission.ORDER_READ,
        Permission.ORDER_WRITE,
        Permission.REPORT_VIEW,
    ],
    "superadmin": [Permission.ADMIN_ALL],
}
def has_permission(role: str, required_permission: Permission) -> bool:
    permissions = ROLE_PERMISSIONS.get(role, [])
    if Permission.ADMIN_ALL in permissions:
        return True
    return required_permission in permissions

# middleware/authorization.py
from functools import wraps
def require_permission(permission: Permission):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            request = kwargs.get("request") or args[0]
            user = request.state.user
            if not has_permission(user.role, permission):
                raise HTTPException(
                    status_code=403,
                    detail=f"Permission denied: {permission.value}"
                )
            return await func(*args, **kwargs)
        return wrapper
    return decorator
# Usage
@router.get("/users")
@require_permission(Permission.USER_READ)
async def list_users(request: Request):
    ...

# models/abac.py
from dataclasses import dataclass
from typing import Any
@dataclass
class PolicyContext:
    subject: dict  # User making the request
    resource: dict  # Resource being accessed
    action: str     # Action being performed
    environment: dict  # Time, IP, etc.
class PolicyEngine:
    def __init__(self):
        self.policies = []
    def add_policy(self, condition, effect="allow"):
        self.policies.append({"condition": condition, "effect": effect})
    def evaluate(self, context: PolicyContext) -> bool:
        for policy in self.policies:
            if policy["condition"](context):
                return policy["effect"] == "allow"
        return False  # Default deny
# Define policies
engine = PolicyEngine()
# Users can only access their own resources
engine.add_policy(
    lambda ctx: ctx.resource.get("owner_id") == ctx.subject.get("id"),
    effect="allow"
)
# Admins can access everything
engine.add_policy(
    lambda ctx: ctx.subject.get("role") == "admin",
    effect="allow"
)
# No access during maintenance window
engine.add_policy(
    lambda ctx: not ctx.environment.get("maintenance_mode", False),
    effect="allow"
)

# models/api_key.py
import secrets
import hashlib
from datetime import datetime
class APIKey:
    def __init__(self, db):
        self.db = db
    async def create(self, user_id: int, name: str, scopes: list[str]) -> dict:
        key = f"ak_{secrets.token_urlsafe(32)}"
        key_hash = hashlib.sha256(key.encode()).hexdigest()