OAuth2 Client Credentials Deep Dive — Server-to-Server Authentication

import jwt, datetime, secrets, hashlib, base64
from flask import Flask, request, jsonify

app = Flask(__name__)
SECRET = "oauth2-secret-key-change-in-production"

# Registered services
clients = {
    "scanner-service": {
        "secret_hash": hashlib.sha256("scanner-secret-123".encode()).hexdigest(),
        "scopes": ["threat:read", "scan:write"],
        "service_name": "Vulnerability Scanner"
    },
    "notification-service": {
        "secret_hash": hashlib.sha256("notify-secret-456".encode()).hexdigest(),
        "scopes": ["alert:write"],
        "service_name": "Alert Dispatcher"
    }
}

@app.route("/oauth/token", methods=["POST"])
def token():
    client_id = request.form.get("client_id")
    client_secret = request.form.get("client_secret")
    grant_type = request.form.get("grant_type")
    scope = request.form.get("scope", "")
    client = clients.get(client_id)

    if grant_type != "client_credentials":
        return jsonify({"error": "unsupported_grant_type"}), 400

    if not client:
        return jsonify({"error": "invalid_client"}), 401

    # Validate client_secret
    secret_hash = hashlib.sha256(client_secret.encode()).hexdigest()
    if secret_hash != client["secret_hash"]:
        return jsonify({"error": "invalid_client"}), 401

    # Validate requested scopes are within allowed scopes
    requested_scopes = scope.split() if scope else client["scopes"]
    allowed_scopes = client["scopes"]
    for s in requested_scopes:
        if s not in allowed_scopes:
            return jsonify({"error": "invalid_scope"}), 400

    token = jwt.encode({
        "iss": "auth.dodatech.com",
        "sub": client_id,
        "aud": "api.dodatech.com",
        "scope": " ".join(requested_scopes),
        "iat": datetime.datetime.utcnow(),
        "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1),
        "jti": secrets.token_hex(16),
        "client_id": client_id
    }, SECRET, algorithm="HS256")

    return jsonify({
        "access_token": token,
        "token_type": "Bearer",
        "expires_in": 3600,
        "scope": " ".join(requested_scopes)
    })

import jwt as pyjwt
from cryptography.hazmat.primitives import serialization

# Client authenticates using a signed JWT instead of client_secret
def validate_private_key_jwt(client_assertion, client_id):
    """Validate client using private_key_jwt client authentication."""
    try:
        # The client's public key must be registered
        client_public_key = get_client_public_key(client_id)

        header = pyjwt.get_unverified_header(client_assertion)
        payload = pyjwt.decode(
            client_assertion,
            client_public_key,
            algorithms=["RS256", "ES256"]
        )

        # Validate JWT contents
        if payload.get("iss") != client_id:
            return False
        if payload.get("sub") != client_id:
            return False
        if "auth.dodatech.com/token" not in payload.get("aud", []):
            return False
        if datetime.datetime.utcfromtimestamp(payload.get("exp", 0)) < datetime.datetime.utcnow():
            return False

        return True
    except Exception:
        return False

@app.route("/oauth/token", methods=["POST"])
def token_with_jwt_auth():
    client_id = request.form.get("client_id")
    client_assertion = request.form.get("client_assertion")
    client_assertion_type = request.form.get("client_assertion_type")

    expected_type = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
    if client_assertion_type != expected_type:
        return jsonify({"error": "invalid_client"}), 401

    if not validate_private_key_jwt(client_assertion, client_id):
        return jsonify({"error": "invalid_client"}), 401

    # Issue token...
    return jsonify({"access_token": "...", "token_type": "Bearer"})

import requests

class ServiceClient:
    """Generic service client using Client Credentials."""

    def __init__(self, auth_url, client_id, client_secret, scopes):
        self.auth_url = auth_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.scopes = scopes
        self.token = None
        self.token_expiry = 0

    def _get_token(self):
        if self.token and time.time() < self.token_expiry - 60:
            return self.token

        resp = requests.post(self.auth_url, data={
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": " ".join(self.scopes)
        })

        data = resp.json()
        self.token = data["access_token"]
        self.token_expiry = time.time() + data["expires_in"]
        return self.token

    def request(self, method, url, **kwargs):
        token = self._get_token()
        headers = kwargs.pop("headers", {})
        headers["Authorization"] = f"Bearer {token}"
        return requests.request(method, url, headers=headers, **kwargs)

# Usage
scanner_client = ServiceClient(
    auth_url="https://auth.dodatech.com/oauth/token",
    client_id="scanner-service",
    client_secret="scanner-secret-123",
    scopes=["threat:read"]
)

response = scanner_client.request("GET", "https://api.dodatech.com/v1/threats")