95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from urllib.parse import urlparse
|
|
|
|
import jwt
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
|
|
|
|
PORT = int(os.environ.get("CASAN_IDP_PORT", "8080"))
|
|
ISSUER = os.environ.get("CASAN_IDP_ISSUER", f"http://127.0.0.1:{PORT}")
|
|
KID = "casan-local-prod-idp"
|
|
KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
|
|
|
|
def b64u_int(value: int) -> str:
|
|
raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
|
|
import base64
|
|
|
|
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
|
|
|
|
|
|
def jwk():
|
|
numbers = KEY.public_key().public_numbers()
|
|
return {
|
|
"kty": "RSA",
|
|
"use": "sig",
|
|
"alg": "RS256",
|
|
"kid": KID,
|
|
"n": b64u_int(numbers.n),
|
|
"e": b64u_int(numbers.e),
|
|
}
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def send_json(self, code, payload):
|
|
body = json.dumps(payload).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self): # noqa: N802
|
|
path = urlparse(self.path).path
|
|
if path == "/healthz":
|
|
self.send_json(200, {"status": "ok", "issuer": ISSUER})
|
|
elif path == "/.well-known/openid-configuration":
|
|
self.send_json(
|
|
200,
|
|
{
|
|
"issuer": ISSUER,
|
|
"jwks_uri": f"{ISSUER}/.well-known/jwks.json",
|
|
"token_endpoint": f"{ISSUER}/token",
|
|
"id_token_signing_alg_values_supported": ["RS256"],
|
|
},
|
|
)
|
|
elif path == "/.well-known/jwks.json":
|
|
self.send_json(200, {"keys": [jwk()]})
|
|
else:
|
|
self.send_json(404, {"error": "not_found"})
|
|
|
|
def do_POST(self): # noqa: N802
|
|
if urlparse(self.path).path != "/token":
|
|
self.send_json(404, {"error": "not_found"})
|
|
return
|
|
size = int(self.headers.get("Content-Length", "0"))
|
|
try:
|
|
payload = json.loads(self.rfile.read(size) or b"{}")
|
|
except json.JSONDecodeError:
|
|
self.send_json(400, {"error": "invalid_json"})
|
|
return
|
|
now = int(time.time())
|
|
claims = {
|
|
"iss": ISSUER,
|
|
"sub": payload.get("sub", "oidc-ops"),
|
|
"role": payload.get("role", "ops"),
|
|
"action": payload.get("action", "deploy"),
|
|
"actor": payload.get("actor", "alice"),
|
|
"input_sha256": payload.get("input_sha256", ""),
|
|
"iat": now,
|
|
"exp": now + int(payload.get("ttl_s", 300)),
|
|
}
|
|
token = jwt.encode(claims, KEY, algorithm="RS256", headers={"kid": KID})
|
|
self.send_json(200, {"access_token": token, "token_type": "Bearer", "expires_in": claims["exp"] - now})
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
HTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
|