Init traefik auth proxy

Author: r-brown

traefik-authproxy/Dockerfile

@@ -0,0 +1,33 @@
+FROM python:3-alpine
+
+# Define build arguments
+ARG USER_ID=1064
+ARG GROUP_ID=1064
+ARG USER_NAME=l64user
+ARG GROUP_NAME=l64group
+
+# Create group and user with the specified IDs
+RUN addgroup -g ${GROUP_ID} ${GROUP_NAME} && \
+    adduser -D -u ${USER_ID} -G ${GROUP_NAME} ${USER_NAME}
+
+# Set working directory
+WORKDIR /home/${USER_NAME}
+
+# Copy the requirements file into the container at the working directory
+COPY --chown=${USER_NAME}:${GROUP_NAME} requirements.txt .
+
+# Copy the main FastAPI application file
+COPY --chown=${USER_NAME}:${GROUP_NAME} traefik_authproxy.py .
+
+# Install any needed packages specified in requirements.txt
+RUN python -m pip install --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Make port 8081 available outside this container
+EXPOSE 8081
+
+# Set user context
+USER ${USER_NAME}
+
+# Run the FastAPI application using Uvicorn
+CMD ["uvicorn", "traefik_authproxy:app", "--host", "0.0.0.0", "--port", "8081"]

traefik-authproxy/README.md

@@ -0,0 +1,46 @@
+<p align="center"><img src="https://raw.githubusercontent.com/Labs64/.github/refs/heads/master/assets/labs64-io-ecosystem.png"></p>
+
+## Traefik Auth (M2M) Middleware
+
+This repository contains a custom Traefik ForwardAuth middleware. The middleware is designed to verify M2M (Machine-to-Machine) JWT tokens issued by Keycloak and enforce path-based role-based access control (RBAC) for microservices deployed on Kubernetes.
+
+It receives incoming requests from Traefik, validates the JWT token, extracts user roles, and checks them against a configurable path/role mapping. If the request is authorized, it allows Traefik to forward the request to the backend service. Otherwise, it returns a `401 Unauthorized` or `403 Forbidden` response.
+
+## Features
+
+- JWT Verification: Validates tokens issued by Keycloak using public keys from the `.well-known` endpoint.
+- Role-Based Access Control (RBAC): Enforces access based on roles assigned to the user/client.
+- Configurable Role Mapping: Allows administrators to define a mapping of URL paths to required roles.
+- FastAPI Backend: A lightweight and performant backend for handling authentication logic.
+
+## Prerequisites
+
+- A running Kubernetes cluster.
+- Traefik installed as an Ingress Controller in your cluster.
+- A configured Keycloak instance.
+- Docker for building the middleware container image.
+
+## Configuration
+
+The middleware is configured using environment variables.
+
+- `KEYCLOAK_URL`: The base URL of your Keycloak instance (e.g., http://keycloak.default.svc.cluster.local:8080).
+- `KEYCLOAK_REALM`: The name of the realm in Keycloak (e.g., labs64io).
+- `KEYCLOAK_AUDIENCE`: The audience claim to verify in the JWT (e.g., labs64io_client).
+- `KEYCLOAK_DISCOVERY_URL`: The URL to the Keycloak discovery endpoint (e.g., http://keycloak.default.svc.cluster.local:8080/realms/labs64io/.well-known/openid-configuration).
+- `ROLE_MAPPING_FILE`: YAML file defining the path-to-role mapping. This can be passed as a ConfigMap in a production environment.
+
+## Usage
+
+- Once deployed, Traefik will intercept any request to *whoami.example.com* and forward it to the auth-middleware for authentication.
+- For a request to be successful, it must include a valid JWT in the Authorization header with the format `Bearer <token>`. The roles contained in the JWT must match the required roles for the requested path as defined in your role mapping.
+- The role mapping is a key part of the middleware's logic. You would define a dictionary that maps a path prefix to a list of required roles.
+
+### For example:
+
+- A request to `/api/admin` would require the `admin` role.
+- A request to `/api/users` would require either the `user` or `admin` role.
+
+## License
+
+This project is licensed under the MIT License.

traefik-authproxy/justfile

@@ -0,0 +1,42 @@
+APP_NAME := "traefik-authproxy"
+
+LOCAL_KEYCLOAK := "http://keycloak.localhost"
+LOCAL_KEYCLOAK_DOCKER := "http://host.docker.internal:8080"
+LOCAL_CLIENT_ID := "labs64io-api-gateway"
+LOCAL_CLIENT_SECRET := "mTEqlt1dDzcVyEOzFjBZV4X8jvEkaQnc"
+
+# build application
+docker:
+    docker build -t {{APP_NAME}}:latest .
+    docker tag {{APP_NAME}}:latest localhost:5005/{{APP_NAME}}:latest
+    docker push localhost:5005/{{APP_NAME}}:latest
+    docker images | grep "{{APP_NAME}}"
+
+# run docker image
+run: docker
+    docker run -p 8081:8081 \
+      -e KEYCLOAK_DISCOVERY_URL="{{LOCAL_KEYCLOAK_DOCKER}}/realms/labs64io/.well-known/openid-configuration" \
+      -e KEYCLOAK_URL="{{LOCAL_KEYCLOAK_DOCKER}}" \
+      -e KEYCLOAK_REALM="labs64io" \
+      -e KEYCLOAK_AUDIENCE="account" \
+      -e ROLE_MAPPING_FILE="/home/l64user/role_mapping.yaml" \
+      -v $(pwd)/sample_role_mapping.yaml:/home/l64user/role_mapping.yaml \
+      {{APP_NAME}}:latest
+
+# open documentation
+docu:
+    open "http://localhost:8081/redoc"
+    open "http://localhost:8081/docs"
+
+
+# open Keycloak well-known configuration
+test-show-wellknown:
+    open "{{LOCAL_KEYCLOAK}}/realms/labs64io/.well-known/openid-configuration"
+
+# generate JWT token
+test-generate-jwt-token:
+    curl --location --request POST '{{LOCAL_KEYCLOAK}}/realms/labs64io/protocol/openid-connect/token' \
+    --header 'Content-Type: application/x-www-form-urlencoded' \
+    --data-urlencode 'grant_type=client_credentials' \
+    --data-urlencode 'client_id={{LOCAL_CLIENT_ID}}' \
+    --data-urlencode 'client_secret={{LOCAL_CLIENT_SECRET}}'

traefik-authproxy/requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.116.1
+uvicorn==0.35.0
+python-jose==3.5.0
+requests==2.32.4
+PyYAML==6.0.2

traefik-authproxy/sample_role_mapping.yaml

@@ -0,0 +1,16 @@
+/api/v1/admin:
+  - admin-role
+
+/api/v1/auditflow:
+  - admin-role
+  - auditflow-role
+  - default-roles-labs64io
+
+/api/v1/ecommerce:
+  - admin-role
+  - ecommerce-role
+
+/public: []
+/v3/api-docs: []
+/actuator: []
+/health: []

traefik-authproxy/traefik_authproxy.py

@@ -0,0 +1,181 @@
+import os
+import logging
+from typing import Dict, Any, List, Tuple
+
+import yaml
+import requests
+from fastapi import FastAPI, Request, HTTPException, status
+from jose import jwt
+from jose.exceptions import JWTError, ExpiredSignatureError
+
+# --- Configuration ---
+KEYCLOAK_URL = os.getenv("KEYCLOAK_URL", "http://keycloak.tools.svc.cluster.local")
+KEYCLOAK_REALM = os.getenv("KEYCLOAK_REALM", "default")
+KEYCLOAK_DISCOVERY_URL = os.getenv(
+    "KEYCLOAK_DISCOVERY_URL",
+    f"{KEYCLOAK_URL}/realms/{KEYCLOAK_REALM}/.well-known/openid-configuration"
+)
+
+KEYCLOAK_AUDIENCE = os.getenv("KEYCLOAK_AUDIENCE", "account")
+ROLE_MAPPING_FILE = os.getenv("ROLE_MAPPING_FILE", "role_mapping.yaml")
+
+# --- Caches ---
+DISCOVERY_CACHE: Dict[str, Any] = {}
+JWKS_CACHE: Dict[str, Any] = {}
+
+# --- Logging ---
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+app_logger = logging.getLogger("forwardauth")
+app_logger.setLevel(logging.DEBUG)
+
+# --- App Initialization ---
+app = FastAPI(
+    title="Traefik Auth (M2M) Middleware",
+    description="ForwardAuth service to verify Keycloak JWTs and enforce RBAC based on URI-to-role mapping",
+    version="1.0.0"
+)
+
+# --- Load Role Mapping and Public Paths ---
+def load_role_mapping(file_path: str) -> Tuple[Dict[str, List[str]], List[str]]:
+    try:
+        with open(file_path, "r") as f:
+            raw_mapping = yaml.safe_load(f)
+
+        if not isinstance(raw_mapping, dict):
+            raise ValueError("Role mapping file must contain a dictionary")
+
+        protected_paths = {}
+        public_paths = []
+
+        for path, roles in raw_mapping.items():
+            if roles in (None, [], ["public"]):
+                public_paths.append(path)
+                app_logger.debug(f"load_role_mapping::Detected public path: {path}")
+            else:
+                protected_paths[path] = roles
+
+        app_logger.info(
+            f"Role mapping loaded: {len(protected_paths)} protected paths, {len(public_paths)} public paths"
+        )
+        return protected_paths, public_paths
+
+    except Exception as e:
+        app_logger.warning(f"load_role_mapping::Skipping path check – failed to load mapping: {e}")
+        return {}, []
+
+PROTECTED_PATHS, PUBLIC_PATHS = load_role_mapping(ROLE_MAPPING_FILE)
+
+# --- JWKS Loader with Discovery ---
+def get_jwks() -> Dict[str, Any]:
+    if JWKS_CACHE:
+        app_logger.debug("get_jwks::Using cached JWKS")
+        return JWKS_CACHE
+
+    try:
+        if "jwks_uri" not in DISCOVERY_CACHE:
+            app_logger.info(f"get_jwks::Fetching discovery doc from {KEYCLOAK_DISCOVERY_URL}")
+            resp = requests.get(KEYCLOAK_DISCOVERY_URL)
+            resp.raise_for_status()
+            jwks_uri = resp.json().get("jwks_uri")
+            if not jwks_uri:
+                raise ValueError("Discovery document missing 'jwks_uri'")
+            DISCOVERY_CACHE["jwks_uri"] = jwks_uri
+
+        jwks_uri = DISCOVERY_CACHE["jwks_uri"]
+        app_logger.info(f"get_jwks::Fetching JWKS from {jwks_uri}")
+        resp = requests.get(jwks_uri)
+        resp.raise_for_status()
+        JWKS_CACHE.update(resp.json())
+        return JWKS_CACHE
+
+    except (requests.RequestException, ValueError) as e:
+        app_logger.error(f"get_jwks::Error: {e}")
+        raise HTTPException(status_code=500, detail="Failed to retrieve JWKS")
+
+# --- JWT Token Verifier ---
+def verify_token(token: str) -> Dict[str, Any]:
+    try:
+        kid = jwt.get_unverified_header(token).get("kid")
+        if not kid:
+            raise HTTPException(status_code=401, detail="Missing 'kid' in token header")
+
+        payload = jwt.decode(
+            token,
+            get_jwks(),
+            algorithms=["RS256"],
+            audience=KEYCLOAK_AUDIENCE
+        )
+        app_logger.debug(f"verify_token::Decoded payload: {payload}")
+        return payload
+
+    except ExpiredSignatureError:
+        raise HTTPException(status_code=401, detail="Token expired")
+    except JWTError as e:
+        raise HTTPException(status_code=401, detail=f"Invalid token: {e}")
+    except Exception as e:
+        app_logger.error("verify_token::Unexpected error", exc_info=True)
+        raise HTTPException(status_code=500, detail="Token verification failed")
+
+# --- Role Extractor ---
+def extract_token_roles(payload: Dict[str, Any]) -> List[str]:
+    roles = set()
+
+    realm_roles = payload.get("realm_access", {}).get("roles", [])
+    if isinstance(realm_roles, list):
+        roles.update(realm_roles)
+
+    client_roles = payload.get("resource_access", {}).get(KEYCLOAK_AUDIENCE, {}).get("roles", [])
+    if isinstance(client_roles, list):
+        roles.update(client_roles)
+
+    return list(roles)
+
+# --- Path Role Matcher ---
+def get_required_roles(path: str) -> List[str]:
+    longest_match = ""
+    required_roles = []
+
+    for prefix, roles in PROTECTED_PATHS.items():
+        if path.startswith(prefix) and len(prefix) > len(longest_match):
+            longest_match = prefix
+            required_roles = roles
+
+    return required_roles
+
+def is_public_path(path: str) -> bool:
+    return any(path.startswith(pub) for pub in PUBLIC_PATHS)
+
+# --- Authentication Endpoint ---
+@app.get("/auth")
+@app.post("/auth")
+async def authenticate(request: Request):
+    forwarded_uri = request.headers.get("X-Forwarded-Uri", "/")
+
+    if is_public_path(forwarded_uri):
+        app_logger.info(f"Public access granted to: {forwarded_uri}")
+        return {"message": "Public access granted", "user_id": None, "roles": []}
+
+    auth_header = request.headers.get("Authorization")
+    if not auth_header or not auth_header.startswith("Bearer "):
+        raise HTTPException(status_code=401, detail="Missing or malformed Authorization header")
+
+    token = auth_header.split(" ")[1]
+    payload = verify_token(token)
+    user_roles = extract_token_roles(payload)
+
+    if not user_roles:
+        raise HTTPException(status_code=403, detail="Token contains no roles")
+
+    required_roles = get_required_roles(forwarded_uri)
+    if not required_roles:
+        raise HTTPException(status_code=403, detail=f"No access control configured for: {forwarded_uri}")
+
+    if not set(user_roles).intersection(required_roles):
+        raise HTTPException(status_code=403, detail=f"Insufficient roles. Required: {required_roles}")
+
+    app_logger.info(f"Access granted to user {payload.get('sub')} for path {forwarded_uri}")
+    return {
+        "message": "Authentication successful",
+        "user_id": payload.get("sub"),
+        "roles": user_roles
+    }