Implementing Zero-Trust Network Access (ZTNA) Concepts in Code

Traditional perimeter security operates on a dangerous assumption: if you're inside the network, you're trusted. This model worked when "inside" meant a physical office with a hardware firewall at the door. Today, with remote work, cloud infrastructure, SaaS tools, and contractor access scattered across the internet, the perimeter has dissolved — and attackers exploit that gap every day. Zero-trust network access (ZTNA) replaces implicit trust with continuous, contextual verification. Here's how to implement its core concepts in actual code.

What Zero Trust Means in Engineering Terms

Vendor marketing has diluted the term, but the underlying principles are concrete and implementable:

1. Never trust, always verify — every request must be authenticated and authorized, regardless of which network it originates from
2. Least privilege — users and services get the minimum access required, scoped to the specific resource they need
3. Assume breach — design your system for the scenario where an attacker already holds valid credentials

These three principles translate to specific engineering choices: short-lived tokens, per-resource policy evaluation, contextual trust scoring, mandatory step-up authentication for sensitive operations, and continuous session validation.

Building an Identity-Aware Middleware in Python

The foundation of ZTNA is moving authentication checks from "once at the perimeter" to "on every single request." Here is a FastAPI middleware that validates JWT tokens and computes a contextual trust score on each call:

import ipaddress
import jwt
from datetime import datetime, timezone
from fastapi import Request, HTTPException

SECRET_KEY = "your-secret-key"
KNOWN_CIDRS = ["10.0.0.0/8", "192.168.0.0/16", "172.16.0.0/12"]

def is_known_network(ip: str) -> bool:
    try:
        addr = ipaddress.ip_address(ip)
        return any(addr in ipaddress.ip_network(cidr) for cidr in KNOWN_CIDRS)
    except ValueError:
        return False

def compute_trust_score(payload: dict, request: Request) -> float:
    score = 1.0

    # Penalize stale tokens — fresh auth signals higher confidence
    issued_at = datetime.fromtimestamp(payload["iat"], tz=timezone.utc)
    age_hours = (datetime.now(timezone.utc) - issued_at).total_seconds() / 3600
    if age_hours > 8:
        score -= 0.5
    elif age_hours > 4:
        score -= 0.3

    # Penalize requests from unrecognized networks
    client_ip = request.client.host
    if not is_known_network(client_ip):
        score -= 0.25

    # Penalize missing user agents (common in automated scanners)
    if not request.headers.get("user-agent", ""):
        score -= 0.2

    return max(score, 0.0)

async def ztna_middleware(request: Request, call_next):
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing bearer token")

    token = auth.split(" ", 1)[1]
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

    trust_score = compute_trust_score(payload, request)
    if trust_score < 0.4:
        raise HTTPException(
            status_code=403,
            detail=f"Request blocked: trust score {trust_score:.2f} below threshold"
        )

    request.state.user = payload
    request.state.trust_score = trust_score
    return await call_next(request)

The trust score approach is more nuanced than binary allow/deny. Different resources can require different minimum scores, and you can tune thresholds per environment without changing your auth infrastructure.

Per-Resource Policy Enforcement

Authentication is necessary but not sufficient. Authorization must be evaluated at the resource level, not at the network edge. Here is a minimal policy engine that enforces role and trust requirements per endpoint:

from dataclasses import dataclass
from typing import Optional

@dataclass
class AccessPolicy:
    resource: str
    allowed_roles: list[str]
    min_trust_score: float = 0.6
    require_recent_auth: bool = False  # token must be < 1 hour old

POLICIES: dict[str, AccessPolicy] = {
    "/api/admin": AccessPolicy(
        resource="/api/admin",
        allowed_roles=["admin"],
        min_trust_score=0.9,
        require_recent_auth=True,
    ),
    "/api/reports": AccessPolicy(
        resource="/api/reports",
        allowed_roles=["admin", "analyst"],
        min_trust_score=0.65,
    ),
    "/api/health": AccessPolicy(
        resource="/api/health",
        allowed_roles=["admin", "analyst", "readonly"],
        min_trust_score=0.2,
    ),
}

def evaluate_access(
    path: str,
    user_role: str,
    trust_score: float,
    token_age_hours: float,
) -> tuple[bool, Optional[str]]:
    policy = POLICIES.get(path)
    if not policy:
        return False, f"No policy defined for '{path}'"

    if user_role not in policy.allowed_roles:
        return False, f"Role '{user_role}' not permitted on '{path}'"

    if trust_score < policy.min_trust_score:
        return False, (
            f"Trust score {trust_score:.2f} below required "
            f"{policy.min_trust_score} for '{path}'"
        )

    if policy.require_recent_auth and token_age_hours > 1.0:
        return False, "Step-up authentication required — please re-authenticate"

    return True, None

The require_recent_auth flag implements step-up authentication: sensitive operations force fresh credentials even when a valid session already exists. This is essential for admin endpoints, write operations on financial data, or any irreversible action.

Short-Lived Tokens and Contextual Binding

Long-lived sessions are fundamentally incompatible with zero trust. If an attacker steals a token, the blast radius is bounded by how long that token remains valid. Enforce short TTLs and embed contextual claims at issuance time:

import secrets
import jwt
from datetime import datetime, timezone, timedelta

SECRET_KEY = "your-secret-key"

def generate_token_id() -> str:
    return secrets.token_hex(16)

def issue_access_token(user_id: str, role: str, client_ip: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "role": role,
        "iat": int(now.timestamp()),
        "exp": int((now + timedelta(minutes=30)).timestamp()),
        "bound_ip": client_ip,      # contextual binding for anomaly detection
        "jti": generate_token_id(), # unique ID for revocation tracking
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

Practical TTL targets: access tokens at 15–30 minutes, refresh tokens as single-use (rotated on every refresh), admin-scoped tokens capped at 15 minutes with no refresh path — require full re-authentication.

The bound_ip claim enables a useful anomaly signal: if a token issued to 10.0.1.15 arrives from 185.220.101.x three minutes later, that warrants investigation even when the signature validates. Store jti values in Redis with a TTL matching the token expiry so you can instantly revoke individual tokens without touching your auth infrastructure.

Continuous Validation: Trust Is Not a One-Time Decision

Zero trust does not stop at request authentication. Sessions need continuous health checks:

• Impossible travel: token issued in Paris, used five minutes later from Singapore
• Behavioral drift: a read-only account issuing bulk DELETE requests
• Device posture changes: certificate revoked or OS vulnerability detected mid-session

Publish auth and access events to a queue — Redis Streams works well for this at moderate scale, Kafka if you are processing millions of events per day. A background worker consumes events, correlates signals, and adds jti values to the revocation list when anomalies cross a threshold.

One anomaly is noise. Three signals within ten minutes warrant automatic session termination. This is where your ZTNA implementation intersects with your SIEM and incident response process. For a structured approach to what events to capture and how to wire alerts, the security hardening checklists we publish cover the event taxonomy and correlation rules used in production deployments.

The Takeaway

Zero trust is not a product you buy — it is a set of engineering patterns applied at every layer. The building blocks are: contextual trust scoring on every request, per-resource policy evaluation, token binding, short TTLs with instant revocation, and anomaly signal correlation.

Start with the gap that matters most in your current architecture. For most teams, that is moving from "authenticated at login" to "validated on every request." The middleware above is a half-day implementation that meaningfully reduces your attack surface. Step-up auth, session revocation, and anomaly detection layer on top of that foundation — and each one closes a different category of attack.

---

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.