API Authentication Capstone Project — Full Auth System Implementation

from flask import Flask, request, jsonify, g
from functools import wraps
import jwt, hashlib, secrets, datetime, pyotp

app = Flask(__name__)
SECRET = secrets.token_hex(32)

# In-memory stores (use Redis/DB in production)
api_keys = {}
refresh_tokens = {}
mfa_secrets = {}

# === MIDDLEWARE ===

def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        # Check API Key (X-API-Key header)
        api_key = request.headers.get("X-API-Key")
        if api_key:
            key_hash = hashlib.sha256(api_key.encode()).hexdigest()
            key_data = api_keys.get(key_hash)
            if key_data and key_data.get("active"):
                g.auth_type = "api_key"
                g.scopes = key_data["scopes"]
                return f(*args, **kwargs)

        # Check Bearer JWT
        auth = request.headers.get("Authorization", "")
        if auth.startswith("Bearer "):
            try:
                payload = jwt.decode(
                    auth[7:], SECRET, algorithms=["HS256"]
                )
                g.auth_type = "jwt"
                g.current_user = payload
                g.scopes = payload.get("scope", "").split()
                return f(*args, **kwargs)
            except jwt.InvalidTokenError:
                pass

        return jsonify({"error": "Authentication required"}), 401
    return decorated

# === API KEY ENDPOINTS ===

@app.route("/api/v1/keys", methods=["POST"])
def create_api_key():
    data = request.get_json()
    raw_key = f"sk-{secrets.token_hex(24)}"
    key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
    api_keys[key_hash] = {
        "name": data.get("name", "default"),
        "scopes": data.get("scopes", ["threat:read"]),
        "active": True,
        "created": datetime.datetime.utcnow().isoformat()
    }
    return jsonify({"api_key": raw_key, "id": key_hash[:8]})

# === JWT ENDPOINTS ===

@app.route("/api/v1/auth/login", methods=["POST"])
def login():
    data = request.get_json()
    # Validate credentials (simplified)
    if data.get("password") != "correct":
        return jsonify({"error": "Invalid credentials"}), 401

    access_token = jwt.encode({
        "sub": data.get("username"),
        "scope": "threat:read threat:write",
        "exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
    }, SECRET, algorithm="HS256")

    refresh = secrets.token_urlsafe(32)
    refresh_tokens[refresh] = {"user": data.get("username")}

    return jsonify({
        "access_token": access_token,
        "refresh_token": refresh,
        "expires_in": 900
    })

# === YOUR TURN: Implement remaining endpoints ===
# 1. POST /api/v1/auth/refresh
# 2. POST /api/v1/auth/logout
# 3. POST /api/v1/oauth/token
# 4. POST /api/v1/mfa/setup
# 5. POST /api/v1/mfa/verify
# 6. Protected threat intelligence endpoints

if __name__ == "__main__":
    app.run(debug=True)

# Test API Key auth
KEY=$(curl -s -X POST -H "Content-Type: application/json" \
  -d '{"name":"test-key","scopes":["threat:read"]}' \
  http://localhost:5000/api/v1/keys | python3 -c "import sys,json;print(json.load(sys.stdin)['api_key'])")

curl -H "X-API-Key: $KEY" http://localhost:5000/api/v1/threats

# Test JWT auth
TOKEN=$(curl -s -X POST -H "Content-Type: application/json" \
  -d '{"username":"admin","password":"correct"}' \
  http://localhost:5000/api/v1/auth/login | python3 -c "import sys,json;print(json.load(sys.stdin)['access_token'])")

curl -H "Authorization: Bearer $TOKEN" http://localhost:5000/api/v1/threats