OAuth2 Token Introspection — RFC 7662 Token Validation for Resource Servers

from flask import Flask, request, jsonify
from datetime import datetime, timezone

app = Flask(__name__)

@app.route('/introspect', methods=['POST'])
def introspect_token():
    """RFC 7662 token introspection endpoint."""
    token = request.form.get('token')
    token_type_hint = request.form.get('token_type_hint', 'access_token')

    # Authenticate the resource server making the request
    client_id, client_secret = authenticate_basic_auth(request)
    if not validate_rs_client(client_id, client_secret):
        return jsonify({'error': 'invalid_client'}), 401

    # Look up the token
    token_data = find_token(token)

    if not token_data or is_expired(token_data):
        return jsonify({'active': False})

    # Return token metadata per RFC 7662
    return jsonify({
        'active': True,
        'sub': token_data['user_id'],
        'client_id': token_data.get('client_id'),
        'scope': ' '.join(token_data.get('scopes', [])),
        'token_type': 'Bearer',
        'exp': int(token_data['expires_at'].timestamp()),
        'iat': int(token_data['issued_at'].timestamp()),
        'iss': 'https://auth.dodatech.com',
        'aud': token_data.get('audience'),
        'jti': token_data.get('token_id'),
        'username': token_data.get('username'),
        'tenant_id': token_data.get('tenant_id'),
        'roles': token_data.get('roles', [])
    })

def is_expired(token_data):
    return token_data['expires_at'] < datetime.now(timezone.utc)

import requests
from cachetools import TTLCache

class IntrospectionClient:
    def __init__(self, introspection_url, client_id, client_secret):
        self.url = introspection_url
        self.auth = (client_id, client_secret)
        self.cache = TTLCache(maxsize=10000, ttl=300)  # 5-min cache

    def introspect(self, token):
        """Introspect a token, using cache when possible."""
        token_hash = hash_token(token)

        if token_hash in self.cache:
            result = self.cache[token_hash]
            if result.get('active'):
                return result
            else:
                # Don't cache inactive results — token may be re-activated
                del self.cache[token_hash]

        # Fresh introspection
        response = requests.post(
            self.url,
            data={'token': token, 'token_type_hint': 'access_token'},
            auth=self.auth
        )
        result = response.json()

        # Cache active results
        if result.get('active'):
            self.cache[token_hash] = result

        return result

    def validate_request(self, request, required_scope=None):
        """Validate the Authorization header and return user info."""
        auth_header = request.headers.get('Authorization', '')
        if not auth_header.startswith('Bearer '):
            return None

        token = auth_header[7:]
        result = self.introspect(token)

        if not result.get('active'):
            return None

        if required_scope:
            token_scopes = result.get('scope', '').split()
            if required_scope not in token_scopes:
                return None

        return {
            'user_id': result['sub'],
            'scopes': result.get('scope', '').split(),
            'tenant_id': result.get('tenant_id'),
            'roles': result.get('roles', [])
        }

# Usage
introspector = IntrospectionClient(
    'https://auth.dodatech.com/introspect',
    'resource-server-1',
    'rs-secret-123'
)

user_info = introspector.validate_request(request, required_scope='read:threats')
if user_info:
    print(f"User {user_info['user_id']} authorized with roles: {user_info['roles']}")
else:
    print("Access denied")

import asyncio
from collections import OrderedDict
import time

class BackgroundRefreshCache:
    def __init__(self, max_size=5000, ttl=300, refresh_before=60):
        self.cache = OrderedDict()
        self.max_size = max_size
        self.ttl = ttl
        self.refresh_before = refresh_before

    def get(self, token_hash, fetcher):
        now = time.time()
        if token_hash in self.cache:
            entry = self.cache[token_hash]
            age = now - entry['cached_at']

            if age < self.ttl:
                # Move to end (most recently used)
                self.cache.move_to_end(token_hash)

                if age > (self.ttl - self.refresh_before):
                    # Trigger background refresh
                    asyncio.ensure_future(self._refresh(token_hash, fetcher))

                return entry['data']

        # Cache miss — fetch synchronously
        return self._fetch_and_cache(token_hash, fetcher, now)

    def _fetch_and_cache(self, token_hash, fetcher, now=None):
        if now is None:
            now = time.time()
        data = fetcher()
        self.cache[token_hash] = {'data': data, 'cached_at': now}
        # Evict oldest if over max size
        while len(self.cache) > self.max_size:
            self.cache.popitem(last=False)
        return data

    async def _refresh(self, token_hash, fetcher):
        """Refresh cache entry in background."""
        try:
            data = await asyncio.to_thread(fetcher)
            if data.get('active'):
                self.cache[token_hash] = {
                    'data': data,
                    'cached_at': time.time()
                }
        except Exception as e:
            print(f"Background refresh failed: {e}")