# WebSocket server — Python (websockets library)
import asyncio, websockets, json

connected = set()

async def handler(websocket):
    connected.add(websocket)
    try:
        async for message in websocket:
            data = json.loads(message)
            # Broadcast to all connected clients
            await asyncio.gather(*[
                ws.send(json.dumps({"user": data["user"], "msg": data["msg"]}))
                for ws in connected if ws != websocket
            ])
    finally:
        connected.remove(websocket)

async def main():
    async with websockets.serve(handler, "localhost", 8765):
        await asyncio.Future()  # run forever

# FastAPI CORS configuration
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],   # never use "*" with credentials
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=86400,
)

import httpx, time

class OAuthClient:
    def __init__(self, token_url, client_id, client_secret, scope):
        self.token_url     = token_url
        self.client_id     = client_id
        self.client_secret = client_secret
        self.scope         = scope
        self._token        = None
        self._expires_at   = 0

    def get_access_token(self) -> str:
        if self._token and time.time() < self._expires_at - 60:  # 60s buffer
            return self._token
        return self._refresh_token()

    def _refresh_token(self) -> str:
        resp = httpx.post(self.token_url, data={
            "grant_type":    "client_credentials",
            "client_id":     self.client_id,
            "client_secret": self.client_secret,
            "scope":         self.scope,
        })
        resp.raise_for_status()
        data = resp.json()
        self._token      = data["access_token"]
        self._expires_at = time.time() + data["expires_in"]
        return self._token

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

import jwt, time
from cryptography.hazmat.primitives import serialization

# ── Verify a JWT (RS256) ───────────────────────────────────
def verify_jwt(token: str, jwks_uri: str) -> dict:
    """Fetch public key from JWKS endpoint and validate token."""
    import httpx
    header = jwt.get_unverified_header(token)
    # Fetch JWKS and find matching key by kid
    jwks = httpx.get(jwks_uri).json()
    public_key = next(k for k in jwks["keys"] if k["kid"] == header["kid"])
    return jwt.decode(
        token,
        jwt.algorithms.RSAAlgorithm.from_jwk(public_key),
        algorithms=["RS256"],
        audience="api.example.com",
        issuer="https://auth.example.com",
        options={"require": ["exp", "iat", "sub", "iss", "aud"]},
    )

# ── Security checks that JWT library handles ──────────────
# ✅ Signature verification
# ✅ Expiry (exp)
# ✅ Not-before (nbf)
# ✅ Issuer (iss)
# ✅ Audience (aud)
# ⚠  Revocation — JWT is stateless; use jti + revocation list in Redis

from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os

# ── AES-256-GCM (symmetric, authenticated encryption) ─────
key   = AESGCM.generate_key(bit_length=256)  # 32 bytes
aesgcm = AESGCM(key)
nonce  = os.urandom(12)          # 96-bit nonce — NEVER reuse with same key
aad    = b"authenticated but not encrypted metadata"

ciphertext = aesgcm.encrypt(nonce, b"secret message", aad)
plaintext  = aesgcm.decrypt(nonce, ciphertext, aad)
# If AAD or ciphertext is tampered → InvalidTag exception (authentication)

# ── RSA encryption/decryption ─────────────────────────────
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes

private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_key  = private_key.public_key()

encrypted = public_key.encrypt(
    b"secret",
    padding.OAEP(mgf=padding.MGF1(hashes.SHA256()), algorithm=hashes.SHA256(), label=None)
)
decrypted = private_key.decrypt(encrypted,
    padding.OAEP(mgf=padding.MGF1(hashes.SHA256()), algorithm=hashes.SHA256(), label=None)
)

# RSA practical limit: encrypt only small data (< key_size - overhead)
# In TLS: RSA/ECDH exchange symmetric session key, then use AES for bulk

from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes, serialization

# Sign a message (private key)
signature = private_key.sign(
    b"document content",
    padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),
    hashes.SHA256()
)

# Verify (public key) — raises InvalidSignature if tampered
try:
    public_key.verify(
        signature,
        b"document content",
        padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),
        hashes.SHA256()
    )
    print("Signature valid")
except Exception:
    print("Tampered!")

import hashlib, hmac, os
import bcrypt
# pip install argon2-cffi
from argon2 import PasswordHasher

# ── SHA-256 for data integrity ─────────────────────────────
digest = hashlib.sha256(b"file contents").hexdigest()

# HMAC — keyed hash (prevents length extension attacks)
mac = hmac.new(b"secret_key", b"message", hashlib.sha256).hexdigest()
# Use hmac.compare_digest() for constant-time comparison (prevent timing attacks)
valid = hmac.compare_digest(mac, received_mac)

# ── bcrypt for passwords ───────────────────────────────────
password = b"user_password_123"
hashed   = bcrypt.hashpw(password, bcrypt.gensalt(rounds=12))  # cost=12
verified = bcrypt.checkpw(password, hashed)  # True

# ── Argon2id (recommended) ─────────────────────────────────
ph = PasswordHasher(
    time_cost=3,       # iterations
    memory_cost=65536, # 64 MB
    parallelism=4,     # threads
)
hashed   = ph.hash("user_password")
try:
    ph.verify(hashed, "user_password")  # True
    # Check if rehash needed (parameters upgraded)
    if ph.check_needs_rehash(hashed):
        hashed = ph.hash("user_password")
except Exception:
    pass  # invalid password