Zero-Trust Implementation Guide

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# Example OAuth 2.0 + OIDC configuration
# Note: This is demonstration code only, not production-ready
auth_config:
  provider: "auth0"
  domain: "your-domain.auth0.com"
  client_id: "${AUTH0_CLIENT_ID}"
  client_secret: "${AUTH0_CLIENT_SECRET}"
  
  scopes:
    - openid
    - profile
    - email
    - read:protected-resource
    
  token_validation:
    algorithm: "RS256"
    issuer: "https://your-domain.auth0.com/"
    audience: "your-api-identifier"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

// JWT token validation middleware
// Note: This is demonstration code only, not production-ready
const jwt = require('jsonwebtoken');
const jwksClient = require('jwks-rsa');

const client = jwksClient({
  jwksUri: 'https://your-domain.auth0.com/.well-known/jwks.json'
});

function getKey(header, callback) {
  client.getSigningKey(header.kid, (err, key) => {
    const signingKey = key.publicKey || key.rsaPublicKey;
    callback(null, signingKey);
  });
}

const validateToken = (req, res, next) => {
  const token = req.headers.authorization?.split(' ')[1];
  
  if (!token) {
    return res.status(401).json({ error: 'No token provided' });
  }
  
  jwt.verify(token, getKey, {
    audience: process.env.AUTH0_AUDIENCE,
    issuer: process.env.AUTH0_ISSUER,
    algorithms: ['RS256']
  }, (err, decoded) => {
    if (err) {
      return res.status(401).json({ error: 'Invalid token' });
    }
    
    req.user = decoded;
    next();
  });
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# Istio service mesh configuration for zero-trust
# Note: This is demonstration code only, not production-ready
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: production
spec:
  mtls:
    mode: STRICT

---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: user-service-policy
  namespace: production
spec:
  selector:
    matchLabels:
      app: user-service
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/production/sa/api-gateway"]
  - to:
    - operation:
        methods: ["GET", "POST"]
        paths: ["/api/users/*"]
  - when:
    - key: custom.jwt_claim
      values: ["admin", "user"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

# FastAPI with zero-trust principles
# Note: This is demonstration code only, not production-ready
from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import httpx
import jwt
from typing import Dict, List

app = FastAPI()
security = HTTPBearer()

class ZeroTrustValidator:
    def __init__(self):
        self.jwks_url = "https://your-domain.auth0.com/.well-known/jwks.json"
        self.issuer = "https://your-domain.auth0.com/"
        self.audience = "your-api-identifier"
    
    async def validate_token(self, token: str) -> Dict:
        try:
            # Fetch JWKS for token validation
            async with httpx.AsyncClient() as client:
                response = await client.get(self.jwks_url)
                jwks = response.json()
            
            # Decode and validate token
            unverified_header = jwt.get_unverified_header(token)
            rsa_key = self._get_rsa_key(jwks, unverified_header['kid'])
            
            payload = jwt.decode(
                token,
                rsa_key,
                algorithms=['RS256'],
                audience=self.audience,
                issuer=self.issuer
            )
            
            return payload
            
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="Token expired")
        except jwt.JWTClaimsError:
            raise HTTPException(status_code=401, detail="Invalid claims")
        except Exception:
            raise HTTPException(status_code=401, detail="Invalid token")
    
    def _get_rsa_key(self, jwks: Dict, kid: str) -> str:
        for key in jwks['keys']:
            if key['kid'] == kid:
                return jwt.algorithms.RSAAlgorithm.from_jwk(key)
        raise HTTPException(status_code=401, detail="Invalid key ID")

validator = ZeroTrustValidator()

async def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    token = credentials.credentials
    payload = await validator.validate_token(token)
    return payload

# Protected endpoint with role-based access
@app.get("/api/sensitive-data")
async def get_sensitive_data(current_user: Dict = Depends(verify_token)):
    # Additional authorization checks
    required_roles = ["admin", "data-reader"]
    user_roles = current_user.get("https://your-app.com/roles", [])
    
    if not any(role in user_roles for role in required_roles):
        raise HTTPException(status_code=403, detail="Insufficient permissions")
    
    return {"data": "sensitive information"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# Row-level security implementation
# Note: This is demonstration code only, not production-ready
class SecureDataAccess:
    def __init__(self, db_connection, user_context):
        self.db = db_connection
        self.user_context = user_context
    
    async def get_user_data(self, user_id: str):
        # Verify user can only access their own data
        if not self._can_access_user_data(user_id):
            raise PermissionError("Access denied")
        
        query = """
        SELECT * FROM user_data 
        WHERE user_id = %s 
        AND (
            user_id = %s OR 
            %s = ANY(SELECT role FROM user_roles WHERE user_id = %s AND role = 'admin')
        )
        """
        
        return await self.db.fetch_all(
            query, 
            [user_id, self.user_context['sub'], self.user_context['sub'], self.user_context['sub']]
        )
    
    def _can_access_user_data(self, requested_user_id: str) -> bool:
        # Same user or admin role
        return (
            self.user_context['sub'] == requested_user_id or
            'admin' in self.user_context.get('https://your-app.com/roles', [])
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# Secure Docker image with non-root user
# Note: This is demonstration code only, not production-ready
FROM python:3.11-slim

# Create non-root user
RUN groupadd -r appuser && useradd -r -g appuser appuser

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY --chown=appuser:appuser . /app
WORKDIR /app

# Switch to non-root user
USER appuser

# Security configurations
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# Network policy for zero-trust networking
# Note: This is demonstration code only, not production-ready
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: zero-trust-policy
  namespace: production
spec:
  podSelector:
    matchLabels:
      app: web-service
  policyTypes:
  - Ingress
  - Egress
  ingress:
  - from:
    - namespaceSelector:
        matchLabels:
          name: api-gateway
    - podSelector:
        matchLabels:
          app: api-gateway
    ports:
    - protocol: TCP
      port: 8000
  egress:
  - to:
    - namespaceSelector:
        matchLabels:
          name: database
    ports:
    - protocol: TCP
      port: 5432

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

# Note: This is demonstration code only, not production-ready
import structlog
from datetime import datetime

logger = structlog.get_logger()

def log_security_event(event_type: str, user_id: str, resource: str, action: str, result: str):
    """Structured logging for security events"""
    logger.info(
        "security_event",
        timestamp=datetime.utcnow().isoformat(),
        event_type=event_type,
        user_id=user_id,
        resource=resource,
        action=action,
        result=result,
        source_ip=request.remote_addr if 'request' in globals() else 'unknown'
    )

# Usage in application
@app.post("/api/update-profile")
async def update_profile(profile_data: dict, current_user: Dict = Depends(verify_token)):
    try:
        # Update logic here
        log_security_event(
            event_type="data_modification",
            user_id=current_user['sub'],
            resource="user_profile",
            action="update",
            result="success"
        )
        return {"status": "updated"}
    except Exception as e:
        log_security_event(
            event_type="data_modification",
            user_id=current_user['sub'],
            resource="user_profile",
            action="update",
            result="failure"
        )
        raise