Authentication Skill

Provides comprehensive authentication and authorization capabilities for the Golden Armada AI Agent Fleet Platform.

When to Use This Skill

Activate this skill when working with:

• User authentication flows
• JWT token management
• OAuth2 integration
• Session management
• Role-based access control (RBAC)

JWT Authentication

Token Generation

```python from jose import jwt from datetime import datetime, timedelta from passlib.context import CryptContext

SECRET_KEY = os.environ["JWT_SECRET"] ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 REFRESH_TOKEN_EXPIRE_DAYS = 7

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(user_id: str, roles: list[str]) -> str: expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) payload = { "sub": user_id, "roles": roles, "exp": expire, "iat": datetime.utcnow(), "type": "access" } return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(user_id: str) -> str: expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS) payload = { "sub": user_id, "exp": expire, "type": "refresh" } return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str) -> dict: try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Token expired") except jwt.JWTError: raise HTTPException(status_code=401, detail="Invalid token") ```

Password Hashing

```python def hash_password(password: str) -> str: return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password)

Usage

async def authenticate_user(email: str, password: str) -> User | None: user = await get_user_by_email(email) if not user: return None if not verify_password(password, user.hashed_password): return None return user ```

FastAPI Auth Dependencies

```python from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User: payload = verify_token(token) user = await get_user(payload["sub"]) if not user: raise HTTPException(status_code=401, detail="User not found") return user

async def get_current_active_user(user: User = Depends(get_current_user)) -> User: if not user.is_active: raise HTTPException(status_code=400, detail="Inactive user") return user

Role-based access

def require_roles(*roles: str): async def role_checker(user: User = Depends(get_current_user)): if not any(role in user.roles for role in roles): raise HTTPException(status_code=403, detail="Insufficient permissions") return user return role_checker

Usage

@app.get("/admin") async def admin_route(user: User = Depends(require_roles("admin"))): return {"message": "Admin access granted"} ```

OAuth2 Integration

Google OAuth2

```python from authlib.integrations.starlette_client import OAuth

oauth = OAuth() oauth.register( name='google', client_id=os.environ['GOOGLE_CLIENT_ID'], client_secret=os.environ['GOOGLE_CLIENT_SECRET'], server_metadata_url='https://accounts.google.com/.well-known/openid-configuration', client_kwargs={'scope': 'openid email profile'} )

@app.get('/auth/google') async def google_login(request: Request): redirect_uri = request.url_for('google_callback') return await oauth.google.authorize_redirect(request, redirect_uri)

@app.get('/auth/google/callback') async def google_callback(request: Request): token = await oauth.google.authorize_access_token(request) user_info = token.get('userinfo')

# Find or create user
user = await get_or_create_user(
    email=user_info['email'],
    name=user_info['name'],
    provider='google'
)

# Generate JWT
access_token = create_access_token(user.id, user.roles)
return {"access_token": access_token, "token_type": "bearer"}

```

GitHub OAuth2

```python oauth.register( name='github', client_id=os.environ['GITHUB_CLIENT_ID'], client_secret=os.environ['GITHUB_CLIENT_SECRET'], authorize_url='https://github.com/login/oauth/authorize', access_token_url='https://github.com/login/oauth/access_token', api_base_url='https://api.github.com/', client_kwargs={'scope': 'user:email'} ) ```

Session Management

```python from fastapi import Request, Response import secrets

async def create_session(user_id: str, response: Response) -> str: session_id = secrets.token_urlsafe(32)

# Store in Redis
await redis.hset(f"session:{session_id}", mapping={
    "user_id": user_id,
    "created_at": datetime.utcnow().isoformat()
})
await redis.expire(f"session:{session_id}", 86400)  # 24 hours

# Set cookie
response.set_cookie(
    key="session_id",
    value=session_id,
    httponly=True,
    secure=True,
    samesite="lax",
    max_age=86400
)

return session_id

async def get_session(request: Request) -> dict | None: session_id = request.cookies.get("session_id") if not session_id: return None

session = await redis.hgetall(f"session:{session_id}")
if not session:
    return None

# Refresh TTL
await redis.expire(f"session:{session_id}", 86400)
return session

async def destroy_session(request: Request, response: Response): session_id = request.cookies.get("session_id") if session_id: await redis.delete(f"session:{session_id}") response.delete_cookie("session_id") ```

RBAC Implementation

```python from enum import Enum from typing import Set

class Permission(str, Enum): READ_AGENTS = "read:agents" WRITE_AGENTS = "write:agents" DELETE_AGENTS = "delete:agents" ADMIN = "admin"

ROLE_PERMISSIONS: dict[str, Set[Permission]] = { "viewer": {Permission.READ_AGENTS}, "operator": {Permission.READ_AGENTS, Permission.WRITE_AGENTS}, "admin": {Permission.READ_AGENTS, Permission.WRITE_AGENTS, Permission.DELETE_AGENTS, Permission.ADMIN}, }

def has_permission(user_roles: list[str], required: Permission) -> bool: for role in user_roles: if role in ROLE_PERMISSIONS and required in ROLE_PERMISSIONS[role]: return True return False

def require_permission(permission: Permission): async def permission_checker(user: User = Depends(get_current_user)): if not has_permission(user.roles, permission): raise HTTPException(status_code=403, detail="Permission denied") return user return permission_checker

Usage

@app.delete("/agents/{id}") async def delete_agent( id: str, user: User = Depends(require_permission(Permission.DELETE_AGENTS)) ): await agent_service.delete(id) return {"status": "deleted"} ```

Security Best Practices

1. Use HTTPS always in production
2. Hash passwords with bcrypt or argon2
3. Short-lived access tokens (15-30 minutes)
4. Refresh token rotation on each use
5. HttpOnly, Secure cookies for tokens
6. Rate limit authentication endpoints
7. Log authentication events for auditing
8. Implement account lockout after failed attempts

OAuth 2.0 & OIDC Best Practices

OAuth 2.0 Grant Types

Authorization Code Flow (Recommended for Web Apps)

from authlib.integrations.starlette_client import OAuth

oauth = OAuth()
oauth.register(
    name='custom_provider',
    client_id=os.environ['OAUTH_CLIENT_ID'],
    client_secret=os.environ['OAUTH_CLIENT_SECRET'],
    authorize_url='https://provider.com/oauth/authorize',
    authorize_params=None,
    access_token_url='https://provider.com/oauth/token',
    access_token_params=None,
    refresh_token_url=None,
    client_kwargs={'scope': 'openid profile email'}
)

@app.get('/auth/login')
async def login(request: Request):
    # Generate state for CSRF protection
    state = secrets.token_urlsafe(32)
    await redis.set(f"oauth_state:{state}", "1", ex=600)

    redirect_uri = request.url_for('auth_callback')
    return await oauth.custom_provider.authorize_redirect(
        request,
        redirect_uri,
        state=state
    )

@app.get('/auth/callback')
async def auth_callback(request: Request):
    # Verify state (CSRF protection)
    state = request.query_params.get('state')
    if not await redis.get(f"oauth_state:{state}"):
        raise HTTPException(status_code=400, detail="Invalid state")

    await redis.delete(f"oauth_state:{state}")

    # Exchange authorization code for tokens
    token = await oauth.custom_provider.authorize_access_token(request)
    user_info = token.get('userinfo')

    # Create or update user
    user = await upsert_user(user_info)

    # Issue application tokens
    access_token = create_access_token(user.id, user.roles)
    refresh_token = create_refresh_token(user.id)

    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

PKCE (Proof Key for Code Exchange) for SPAs

# Frontend (JavaScript/TypeScript)
import { generateCodeVerifier, generateCodeChallenge } from 'oauth-pkce'

// Generate PKCE parameters
const codeVerifier = generateCodeVerifier()
const codeChallenge = await generateCodeChallenge(codeVerifier)

// Store verifier for later use
sessionStorage.setItem('code_verifier', codeVerifier)

// Redirect to authorization endpoint
const authUrl = new URL('https://provider.com/oauth/authorize')
authUrl.searchParams.set('client_id', CLIENT_ID)
authUrl.searchParams.set('redirect_uri', REDIRECT_URI)
authUrl.searchParams.set('response_type', 'code')
authUrl.searchParams.set('scope', 'openid profile email')
authUrl.searchParams.set('code_challenge', codeChallenge)
authUrl.searchParams.set('code_challenge_method', 'S256')
authUrl.searchParams.set('state', generateState())

window.location.href = authUrl.toString()

// In callback handler
const code = new URLSearchParams(window.location.search).get('code')
const codeVerifier = sessionStorage.getItem('code_verifier')

const tokenResponse = await fetch('https://provider.com/oauth/token', {
  method: 'POST',
  headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
  body: new URLSearchParams({
    grant_type: 'authorization_code',
    code: code,
    redirect_uri: REDIRECT_URI,
    client_id: CLIENT_ID,
    code_verifier: codeVerifier
  })
})

Client Credentials Flow (Service-to-Service)

import httpx
from datetime import datetime, timedelta

class ServiceAuthClient:
    def __init__(self):
        self.token = None
        self.expires_at = None

    async def get_token(self) -> str:
        # Return cached token if still valid
        if self.token and self.expires_at > datetime.utcnow():
            return self.token

        # Request new token
        async with httpx.AsyncClient() as client:
            response = await client.post(
                'https://provider.com/oauth/token',
                data={
                    'grant_type': 'client_credentials',
                    'client_id': os.environ['SERVICE_CLIENT_ID'],
                    'client_secret': os.environ['SERVICE_CLIENT_SECRET'],
                    'scope': 'api.read api.write'
                }
            )
            response.raise_for_status()
            data = response.json()

            self.token = data['access_token']
            self.expires_at = datetime.utcnow() + timedelta(seconds=data['expires_in'] - 60)

            return self.token

# Usage
auth_client = ServiceAuthClient()

async def call_protected_api():
    token = await auth_client.get_token()
    async with httpx.AsyncClient() as client:
        response = await client.get(
            'https://api.service.com/resource',
            headers={'Authorization': f'Bearer {token}'}
        )
        return response.json()

OpenID Connect (OIDC)

ID Token Validation

from jose import jwt, jwk
from jose.utils import base64url_decode
import httpx

class OIDCValidator:
    def __init__(self, issuer: str, client_id: str):
        self.issuer = issuer
        self.client_id = client_id
        self.jwks = None
        self.jwks_updated_at = None

    async def get_jwks(self) -> dict:
        # Refresh JWKS if stale (cache for 24 hours)
        if not self.jwks or (datetime.utcnow() - self.jwks_updated_at).seconds > 86400:
            async with httpx.AsyncClient() as client:
                response = await client.get(f"{self.issuer}/.well-known/jwks.json")
                response.raise_for_status()
                self.jwks = response.json()
                self.jwks_updated_at = datetime.utcnow()

        return self.jwks

    async def validate_id_token(self, id_token: str) -> dict:
        # Decode header to get key ID
        header = jwt.get_unverified_header(id_token)
        kid = header['kid']

        # Get JWKS and find matching key
        jwks = await self.get_jwks()
        key = next((k for k in jwks['keys'] if k['kid'] == kid), None)

        if not key:
            raise ValueError("Public key not found in JWKS")

        # Convert JWK to PEM
        public_key = jwk.construct(key)

        # Validate and decode ID token
        try:
            claims = jwt.decode(
                id_token,
                public_key.to_pem().decode('utf-8'),
                algorithms=['RS256'],
                audience=self.client_id,
                issuer=self.issuer,
                options={
                    'verify_exp': True,
                    'verify_iat': True,
                    'verify_aud': True,
                    'verify_iss': True
                }
            )

            # Additional validations
            if claims.get('nonce'):
                # Verify nonce matches what was sent in auth request
                pass

            return claims

        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="ID token expired")
        except jwt.JWTClaimsError as e:
            raise HTTPException(status_code=401, detail=f"Invalid ID token claims: {e}")
        except Exception as e:
            raise HTTPException(status_code=401, detail=f"ID token validation failed: {e}")

# Usage
validator = OIDCValidator(
    issuer="https://provider.com",
    client_id=os.environ['OIDC_CLIENT_ID']
)

@app.post("/auth/oidc/callback")
async def oidc_callback(id_token: str):
    claims = await validator.validate_id_token(id_token)

    # Extract user information
    user = await get_or_create_user(
        email=claims['email'],
        name=claims['name'],
        sub=claims['sub']
    )

    return {"user": user, "claims": claims}

JWT Security Best Practices

Secure Token Generation

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
import os

# Use RS256 (asymmetric) instead of HS256 for public verification
PRIVATE_KEY_PATH = os.environ.get('JWT_PRIVATE_KEY_PATH')
PUBLIC_KEY_PATH = os.environ.get('JWT_PUBLIC_KEY_PATH')

def load_keys():
    with open(PRIVATE_KEY_PATH, 'rb') as f:
        private_key = serialization.load_pem_private_key(
            f.read(),
            password=None,
            backend=default_backend()
        )

    with open(PUBLIC_KEY_PATH, 'rb') as f:
        public_key = serialization.load_pem_public_key(
            f.read(),
            backend=default_backend()
        )

    return private_key, public_key

PRIVATE_KEY, PUBLIC_KEY = load_keys()

def create_secure_access_token(
    user_id: str,
    roles: list[str],
    tenant_id: str = None,
    custom_claims: dict = None
) -> str:
    now = datetime.utcnow()

    payload = {
        # Standard claims (RFC 7519)
        "iss": "https://api.yourdomain.com",  # Issuer
        "sub": user_id,                        # Subject (user ID)
        "aud": ["https://api.yourdomain.com"], # Audience
        "exp": now + timedelta(minutes=15),    # Expiration (short-lived)
        "nbf": now,                             # Not before
        "iat": now,                             # Issued at
        "jti": secrets.token_urlsafe(16),      # JWT ID (unique token identifier)

        # Custom claims
        "roles": roles,
        "type": "access",
    }

    # Add tenant context for multi-tenant applications
    if tenant_id:
        payload["tenant_id"] = tenant_id

    # Add any custom claims
    if custom_claims:
        payload.update(custom_claims)

    return jwt.encode(payload, PRIVATE_KEY, algorithm="RS256")

def verify_secure_token(token: str) -> dict:
    try:
        payload = jwt.decode(
            token,
            PUBLIC_KEY,
            algorithms=["RS256"],
            audience=["https://api.yourdomain.com"],
            issuer="https://api.yourdomain.com",
            options={
                'verify_exp': True,
                'verify_nbf': True,
                'verify_iat': True,
                'verify_aud': True,
                'verify_iss': True,
                'require_exp': True,
                'require_iat': True,
                'require_nbf': True
            }
        )

        # Validate token type
        if payload.get('type') != 'access':
            raise HTTPException(status_code=401, detail="Invalid token type")

        return payload

    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError as e:
        raise HTTPException(status_code=401, detail=f"Invalid token: {e}")

Token Revocation with Blacklist

class TokenBlacklist:
    """Redis-based token blacklist for revoked tokens"""

    def __init__(self, redis_client):
        self.redis = redis_client

    async def revoke_token(self, jti: str, exp: int):
        """Add token to blacklist until expiration"""
        ttl = exp - int(datetime.utcnow().timestamp())
        if ttl > 0:
            await self.redis.set(f"blacklist:{jti}", "1", ex=ttl)

    async def is_revoked(self, jti: str) -> bool:
        """Check if token is blacklisted"""
        return await self.redis.exists(f"blacklist:{jti}")

# Global blacklist instance
token_blacklist = TokenBlacklist(redis)

async def get_current_user_with_revocation_check(
    token: str = Depends(oauth2_scheme)
) -> User:
    payload = verify_secure_token(token)

    # Check if token has been revoked
    if await token_blacklist.is_revoked(payload['jti']):
        raise HTTPException(status_code=401, detail="Token has been revoked")

    user = await get_user(payload["sub"])
    if not user:
        raise HTTPException(status_code=401, detail="User not found")

    return user

@app.post("/auth/logout")
async def logout(token: str = Depends(oauth2_scheme)):
    payload = verify_secure_token(token)
    await token_blacklist.revoke_token(payload['jti'], payload['exp'])
    return {"message": "Logged out successfully"}

Security Analysis with Extended Thinking

When reviewing authentication flows, use extended thinking for comprehensive security analysis.

Authentication Flow Security Review Template

## Authentication Flow Security Review

**Flow**: [Login/OAuth/SSO/API Authentication]
**Date**: [YYYY-MM-DD]
**Reviewer**: [Name/Agent]

### Flow Diagram
[Document the authentication flow step-by-step]

### Security Analysis Checklist

#### Confidentiality
- [ ] Credentials transmitted over HTTPS only
- [ ] Passwords hashed with strong algorithm (bcrypt/argon2)
- [ ] Tokens encrypted in transit
- [ ] Sensitive data not logged
- [ ] Secrets stored securely (env vars, secrets manager)

#### Integrity
- [ ] CSRF protection implemented
- [ ] Request tampering prevented
- [ ] Token signature validation
- [ ] State parameter validated (OAuth)
- [ ] Nonce validated (OIDC)

#### Availability
- [ ] Rate limiting on auth endpoints
- [ ] Account lockout after failed attempts
- [ ] DDoS protection in place
- [ ] Graceful degradation strategy
- [ ] Session timeout configured

#### Authentication
- [ ] MFA available for sensitive accounts
- [ ] Password complexity requirements
- [ ] Credential stuffing protection
- [ ] Brute force mitigation
- [ ] Session fixation prevention

#### Authorization
- [ ] Principle of least privilege
- [ ] Role-based access control
- [ ] Permission checks on every request
- [ ] Token scope validation
- [ ] Tenant isolation (multi-tenant apps)

#### Session Management
- [ ] Secure session tokens
- [ ] HttpOnly, Secure, SameSite cookies
- [ ] Session timeout implemented
- [ ] Logout functionality
- [ ] Concurrent session handling

### Extended Thinking Analysis

Use Claude with extended thinking for deep security review:

```python
import anthropic

client = anthropic.Anthropic()

security_review_prompt = """
Perform a comprehensive security analysis of this authentication flow:

[Paste authentication code/flow description]

Analyze for:
1. OWASP Top 10 vulnerabilities
2. Authentication bypass possibilities
3. Token security weaknesses
4. Session management issues
5. Input validation gaps
6. Race conditions
7. Logic flaws

Provide specific findings with:
- Severity (Critical/High/Medium/Low)
- Location (file:line)
- Attack vector
- Remediation steps
"""

response = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=32000,
    thinking={"type": "adaptive"},
    output_config={"effort": "high"},  # deep reasoning for security analysis
    messages=[{
        "role": "user",
        "content": security_review_prompt
    }]
)

# Extract thinking and analysis
for block in response.content:
    if block.type == "thinking":
        print(f"Deep Analysis:\n{block.thinking}\n")
    elif block.type == "text":
        print(f"Findings:\n{block.text}")

Threat Modeling for Authentication

STRIDE Threat Model for Auth Systems

Adapted from [[deep-analysis]] skill threat modeling template:

## Authentication Threat Model

### System: [Auth System Name]
**Version**: 1.0
**Last Updated**: [Date]

### Trust Boundaries

┌─────────────────────────────────────────┐ │ External (Untrusted) │ │ [End Users] [Credential Stuffers] │ │ [MITM Attackers] │ └──────────────────┬──────────────────────┘ │ TLS/HTTPS ┌──────────────────┴──────────────────────┐ │ Public API (Semi-trusted) │ │ [API Gateway] [Auth Endpoints] │ │ [OAuth Providers] [OIDC IdP] │ └──────────────────┬──────────────────────┘ │ Internal Auth ┌──────────────────┴──────────────────────┐ │ Application Layer (Trusted) │ │ [Business Logic] [User Management] │ └──────────────────┬──────────────────────┘ │ DB Protocol ┌──────────────────┴──────────────────────┐ │ Data Layer (Highly Trusted) │ │ [User DB] [Session Store] [Secrets] │ └─────────────────────────────────────────┘

### STRIDE Analysis