Multi System Sso Authentication

Multi-System SSO Authentication Skill

Overview

This skill provides comprehensive patterns for implementing enterprise SSO authentication that supports multiple identity providers. It covers JWT RS256 token validation, backwards verification with authoritative systems, Laravel session decryption, permission mapping, and Redis session management.

When to Use This Skill

• Integrating with multiple enterprise SSO systems
• Implementing secure JWT token validation with backwards verification
• Supporting legacy session-based authentication alongside JWT
• Building unified authentication adapters for microservices
• Mapping permissions across different systems
• Implementing token introspection and revocation
• Handling OAuth2 flows with multiple providers

Core Concepts

Authentication Architecture

┌─────────────────────────────────────────────────────────┐
│                     Your Application                     │
│  ┌────────────────────────────────────────────────────┐ │
│  │          UnifiedAuthAdapter (Router)               │ │
│  │  ┌──────────────────────────────────────────────┐ │ │
│  │  │  Check token issuer (iss claim)              │ │ │
│  │  │  Route to appropriate adapter                │ │ │
│  │  └──────────────────────────────────────────────┘ │ │
│  │         ▼          ▼          ▼          ▼        │ │
│  │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│  │  │  CORP   │ │   SGF   │ │   GED   │ │ CARRINHO│ │ │
│  │  │ Adapter │ │ Adapter │ │ Adapter │ │ Adapter │ │ │
│  │  └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │
│  └────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
         │             │             │             │
         ▼             ▼             ▼             ▼
  ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐
  │ Corporativo│ │    SGF    │ │    GED    │ │ Carrinho  │
  │    SSO    │ │    API    │ │    API    │ │    API    │
  └───────────┘ └───────────┘ └───────────┘ └───────────┘

Token Flow

1. User authenticates with external SSO system
2. SSO system issues JWT with issuer (iss) and audience (aud) claims
3. Your app receives token from request headers
4. UnifiedAuthAdapter routes to appropriate adapter based on issuer
5. Adapter validates JWT signature with public key
6. Backwards verification checks token validity with issuing system
7. Permissions mapped from SSO format to your app's format
8. User session created in Redis for future requests

Project Structure

src/
├── api/
│   ├── middlewares/
│   │   └── auth.py              # AuthMiddleware
│   └── path/
│       └── auth.py              # Authentication endpoints
├── domain/
│   └── modules/
│       └── auth/
│           ├── entity.py        # User entity
│           ├── session.py       # Session management
│           └── permissions.py   # Permission definitions
└── infra/
    ├── adapters/
    │   └── auth/
    │       ├── unified_adapter.py      # Router for all adapters
    │       ├── corporativo_adapter.py  # Corporativo SSO
    │       ├── sgf_adapter.py          # SGF integration
    │       ├── ged_adapter.py          # GED integration
    │       └── carrinho_adapter.py     # Carrinho integration
    ├── cache/
    │   └── redis_session.py     # Redis session storage
    └── services/
        └── permission_mapper.py # Permission mapping

Implementation Patterns

1. Unified Authentication Adapter (Router)

# src/infra/adapters/auth/unified_adapter.py
from typing import Dict, Any
from jose import jwt, JWTError

from src.infra.adapters.auth.corporativo_adapter import CorporativoAuthAdapter
from src.infra.adapters.auth.sgf_adapter import SGFAuthAdapter
from src.infra.adapters.auth.ged_adapter import GEDAuthAdapter
from src.infra.adapters.auth.carrinho_adapter import CarrinhoAuthAdapter
from src.config.settings import app_settings

class UnifiedAuthAdapter:
    """Unified authentication adapter that routes tokens to appropriate SSO adapter.

    Routes based on JWT issuer claim (iss).
    """

    def __init__(
        self,
        corporativo_adapter: CorporativoAuthAdapter,
        sgf_adapter: SGFAuthAdapter,
        ged_adapter: GEDAuthAdapter,
        carrinho_adapter: CarrinhoAuthAdapter,
    ):
        self.adapters = {
            "corporativo": corporativo_adapter,
            "sgf": sgf_adapter,
            "ged": ged_adapter,
            "carrinho": carrinho_adapter,
        }

        # Map issuer URLs to adapter names
        self.issuer_map = {
            app_settings.CORPORATIVO_API_URL: "corporativo",
            app_settings.SGF_API_URL: "sgf",
            app_settings.GED_API_URL: "ged",
            app_settings.CARRINHO_API_URL: "carrinho",
            "gefin-backend": "corporativo",  # Self-issued tokens
        }

    async def validate_token(self, token: str) -> Dict[str, Any]:
        """Validate token and route to appropriate adapter.

        Args:
            token: JWT token string

        Returns:
            User data dictionary with permissions

        Raises:
            JWTError: If token is invalid or from unknown issuer
        """
        # Decode without verification to check issuer
        try:
            unverified = jwt.get_unverified_claims(token)
            issuer = unverified.get("iss")
        except JWTError as e:
            raise JWTError(f"Invalid JWT format: {e}")

        # Map issuer to adapter
        adapter_name = self.issuer_map.get(issuer)
        if not adapter_name:
            raise JWTError(f"Unknown token issuer: {issuer}")

        # Check if adapter is enabled
        enabled_systems = app_settings.ENABLED_AUTH_SYSTEMS
        if adapter_name not in enabled_systems:
            raise JWTError(f"Authentication system '{adapter_name}' is disabled")

        # Route to appropriate adapter
        adapter = self.adapters[adapter_name]
        return await adapter.validate_token(token)

    async def validate_session(self, session_id: str) -> Dict[str, Any]:
        """Validate session cookie (for legacy systems).

        Routes to Corporativo adapter (primary session provider).
        """
        return await self.adapters["corporativo"].validate_session(session_id)

2. Base Auth Adapter Pattern

# src/infra/adapters/auth/base_adapter.py
from abc import ABC, abstractmethod
from typing import Dict, Any

class IAuthAdapter(ABC):
    """Abstract base class for authentication adapters.

    All SSO adapters must implement this interface.
    """

    @abstractmethod
    async def validate_token(self, token: str) -> Dict[str, Any]:
        """Validate JWT token and return user data.

        Args:
            token: JWT token string

        Returns:
            User data with permissions

        Raises:
            JWTError: If token is invalid
        """
        pass

    @abstractmethod
    async def validate_session(self, session_id: str) -> Dict[str, Any]:
        """Validate session ID and return user data.

        Args:
            session_id: Session identifier

        Returns:
            User data with permissions

        Raises:
            SessionError: If session is invalid
        """
        pass

    @abstractmethod
    def get_permissions(self, user_data: Dict[str, Any]) -> list[str]:
        """Extract and map permissions from user data.

        Args:
            user_data: User data from SSO system

        Returns:
            List of permission strings in app format
        """
        pass

3. JWT RS256 Token Validation with Backwards Verification

# src/infra/adapters/auth/corporativo_adapter.py
import httpx
from datetime import datetime, timedelta
from jose import jwt, JWTError

from src.infra.adapters.auth.base_adapter import IAuthAdapter
from src.infra.cache.redis_session import RedisSessionManager

class CorporativoAuthAdapter(IAuthAdapter):
    """Corporativo SSO authentication adapter.

    Implements JWT RS256 validation with backwards verification.
    """

    def __init__(
        self,
        public_key: str,
        private_key: str,
        api_url: str,
        session_manager: RedisSessionManager,
    ):
        self.public_key = public_key
        self.private_key = private_key
        self.api_url = api_url
        self.session_manager = session_manager
        self._validation_cache: Dict[str, tuple[Dict, datetime]] = {}
        self._cache_ttl = 30  # 30 seconds

    async def validate_token(self, token: str) -> Dict[str, Any]:
        """Validate JWT token with backwards verification.

        Steps:
        1. Verify JWT signature with RSA public key
        2. Check issuer and audience claims
        3. Perform backwards verification with SSO system
        4. Map permissions to app format
        """
        try:
            # Verify signature and decode token
            payload = jwt.decode(
                token,
                self.public_key,
                algorithms=["RS256"],
                options={"verify_iss": False, "verify_aud": False},  # Manual validation
            )

            # Manual issuer validation
            accepted_issuers = ["gefin-backend", self.api_url]
            if payload.get("iss") not in accepted_issuers:
                raise JWTError(f"Invalid issuer: {payload.get('iss')}")

            # Manual audience validation
            accepted_audiences = ["gefin-api", "gefin"]
            aud = payload.get("aud")
            if isinstance(aud, list):
                if not any(a in accepted_audiences for a in aud):
                    raise JWTError(f"Invalid audience: {aud}")
            elif aud not in accepted_audiences:
                raise JWTError(f"Invalid audience: {aud}")

            # Check expiration
            exp = payload.get("exp")
            if exp and datetime.fromtimestamp(exp)  None:
        """Verify token validity with Corporativo SSO system.

        Implements backwards verification with caching.
        """
        # Check cache first
        cache_key = payload.get("sub")
        if cache_key in self._validation_cache:
            cached_data, cached_at = self._validation_cache[cache_key]
            if datetime.now() - cached_at  list[str]:
        """Map Corporativo permissions to app format.

        Example mapping:
            "Ver anuidade" -> "gefin.boleto.read"
            "Editar anuidade" -> "gefin.boleto.write"
        """
        corporativo_permissions = user_data.get("permissions", [])
        permission_map = {
            "Ver anuidade": "gefin.boleto.read",
            "Editar anuidade": "gefin.boleto.write",
            "Ver parcelamento": "gefin.parcela.read",
            "Editar parcelamento": "gefin.parcela.write",
            "Ver publicações": "gefin.publicacao.read",
            "Editar publicações": "gefin.publicacao.write",
            # ... more mappings
        }

        mapped_permissions = []
        for corp_perm in corporativo_permissions:
            if corp_perm == "*":  # Admin wildcard
                return ["*"]
            app_perm = permission_map.get(corp_perm)
            if app_perm:
                mapped_permissions.append(app_perm)

        # Ensure at least read permission
        if not any(p.endswith(".read") for p in mapped_permissions):
            mapped_permissions.append("gefin.user.read")

        return mapped_permissions

    async def validate_session(self, session_id: str) -> Dict[str, Any]:
        """Validate session from Redis.

        Falls back to Laravel session decryption if Redis unavailable.
        """
        # Try Redis first
        session_data = await self.session_manager.get_session(session_id)
        if session_data:
            return session_data

        # Fall back to Laravel session decryption
        return await self._decrypt_laravel_session(session_id)

    async def _decrypt_laravel_session(self, session_cookie: str) -> Dict[str, Any]:
        """Decrypt Laravel AES-256-CBC session cookie.

        Laravel session format:
        - base64(iv:encrypted_payload:mac)
        - Encrypted with APP_KEY from .env
        """
        # Implementation omitted for brevity
        # See Laravel session decryption pattern below
        pass

4. Laravel Session Decryption

# src/infra/adapters/auth/laravel_session.py
import base64
import json
import hashlib
import hmac
from Cryptodome.Cipher import AES
from Cryptodome.Util.Padding import unpad
import phpserialize

class LaravelSessionDecryptor:
    """Decrypt Laravel AES-256-CBC encrypted sessions.

    Handles Laravel's session encryption format.
    """

    def __init__(self, app_key: str):
        """Initialize with Laravel APP_KEY.

        Args:
            app_key: Laravel APP_KEY from .env (base64: prefix)
        """
        # Remove 'base64:' prefix if present
        if app_key.startswith("base64:"):
            app_key = app_key[7:]

        self.key = base64.b64decode(app_key)

    def decrypt(self, encrypted_value: str) -> str:
        """Decrypt Laravel encrypted value.

        Format: base64(json({"iv": "...", "value": "...", "mac": "..."}))
        """
        # Decode base64
        decoded = base64.b64decode(encrypted_value)
        payload = json.loads(decoded)

        # Verify MAC signature
        if not self._valid_mac(payload):
            raise ValueError("Invalid MAC signature")

        # Decrypt
        iv = base64.b64decode(payload["iv"])
        encrypted = base64.b64decode(payload["value"])

        cipher = AES.new(self.key, AES.MODE_CBC, iv)
        decrypted = unpad(cipher.decrypt(encrypted), AES.block_size)

        return decrypted.decode("utf-8")

    def _valid_mac(self, payload: dict) -> bool:
        """Verify MAC signature."""
        mac = payload.get("mac")
        if not mac:
            return False

        # Calculate expected MAC
        message = base64.b64encode(
            json.dumps({"iv": payload["iv"], "value": payload["value"]}).encode()
        )
        expected_mac = hmac.new(
            self.key,
            message,
            hashlib.sha256,
        ).hexdigest()

        return hmac.compare_digest(mac, expected_mac)

    def decrypt_session(self, session_cookie: str) -> dict:
        """Decrypt Laravel session cookie and extract user data.

        Args:
            session_cookie: Laravel session cookie value

        Returns:
            Dictionary with user_id and other session data
        """
        # Decrypt session
        decrypted = self.decrypt(session_cookie)

        # Unserialize PHP session data
        session_data = phpserialize.loads(decrypted.encode())

        # Extract user ID from various Laravel guard patterns
        user_id = None

        # Pattern 1: login_web_{guard}_*
        for key in session_data:
            if isinstance(key, bytes):
                key_str = key.decode()
                if key_str.startswith("login_web_"):
                    user_id = session_data[key]
                    break

        # Pattern 2: Direct user_id key
        if not user_id and b"user_id" in session_data:
            user_id = session_data[b"user_id"]

        if not user_id:
            raise ValueError("No user_id found in session")

        return {
            "user_id": user_id.decode() if isinstance(user_id, bytes) else user_id,
            "session_data": session_data,
        }

5. Redis Session Management

# src/infra/cache/redis_session.py
import json
from datetime import timedelta
from redis.asyncio import Redis

class RedisSessionManager:
    """Manage user sessions in Redis.

    Stores session data with TTL for automatic expiration.
    """

    def __init__(self, redis_client: Redis, ttl_seconds: int = 28800):
        """Initialize session manager.

        Args:
            redis_client: Async Redis client
            ttl_seconds: Session TTL (defa

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [rafaelkamimura](https://github.com/rafaelkamimura)
- **Source:** [rafaelkamimura/claude-tools](https://github.com/rafaelkamimura/claude-tools)
- **License:** MIT

Install and usage instructions live in the source repository linked above.