Compare commits

...

4 Commits

Author SHA1 Message Date
df82068aeb fix: CSRF protection, cookie security flags, rate limiting, XSS (#23,#24,#25,#28,#33)
Add CSRF tokens to all cookie-based POST endpoints.
Set Secure and SameSite=Strict on auth cookie.
Rate limit login to 5 attempts per 15min per IP.
Escape HTML in signing error messages (XSS fix).
Remove duplicate get_user_from_cookie definition.
2026-07-04 04:51:56 +00:00
c759597ad0 fix: pass YubiKey PIN via stdin instead of CLI arg (#18, #31)
--pin-source stdin prevents PIN visibility in `ps` output.
Use tempfile.mkstemp for all temp files (unpredictable names, 0600 perms).
Clean up temp files in finally block.
Add tests for PIN not in args and mkstemp usage.
2026-07-04 04:49:56 +00:00
279a28515f fix: generate JWT secret on first boot instead of hardcoded default (#19)
Generate random 256-bit secret stored in /var/lib/certauth/.jwt_secret (mode 0600)
ENV JWT_SECRET takes precedence. Remove CHANGE_ME_JWT_SECRET default.
Add tests for secret generation and env override
2026-07-04 04:48:30 +00:00
c4d952430e update local changes 2026-07-03 01:14:04 +00:00
12 changed files with 446 additions and 66 deletions

15
Caddyfile-playground.ms Normal file
View File

@ -0,0 +1,15 @@
{
admin off
}
example.com, www.example.com {
encode gzip
tls /etc/ssl/certs/example.com.pem /etc/ssl/private/example.com.key
root * /var/www/example.com
file_server browse
}
:80 {
redir https://{host}{uri} permanent
}

View File

@ -1,18 +1,63 @@
import os
import secrets
YK_ROOT_SERIAL = "35450561"
YK_ROOT_PIN = os.environ.get("YK_ROOT_PIN", "CHANGE_ME_YK1_PIN")
YK_INT_SERIAL = "33930436"
YK_INT_PIN = os.environ.get("YK_INT_PIN", "CHANGE_ME_YK2_PIN")
ROOT_CA_PATH = "/etc/ssl/ca/root/root-ca.crt"
INT_CA_PATH = "/etc/ssl/ca/intermediate/intermediate-ca.crt"
CA_CHAIN_PATH = "/etc/ssl/ca/ca-chain.crt"
ISSUED_DIR = "/etc/ssl/ca/issued"
DB_PATH = "/var/lib/certauth/certauth.db"
SECRET_KEY = os.environ.get("JWT_SECRET", "CHANGE_ME_JWT_SECRET")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
ADMIN_USERNAME = "certauth"
PKCS11_MODULE = "/usr/lib/aarch64-linux-gnu/opensc-pkcs11.so"
YK_PUB_ROOT = "/tmp/yk1-root-pub.pem"
YK_PUB_INT = "/tmp/yk2-int-pub.pem"
_YK_ROOT_SERIAL = "35450561"
_YK_INT_SERIAL = "33930436"
_ROOT_CA_PATH = "/etc/ssl/ca/root/root-ca.crt"
_INT_CA_PATH = "/etc/ssl/ca/intermediate/intermediate-ca.crt"
_CA_CHAIN_PATH = "/etc/ssl/ca/ca-chain.crt"
_ISSUED_DIR = "/etc/ssl/ca/issued"
_DB_PATH = "/var/lib/certauth/certauth.db"
_JWT_SECRET_FILE = "/var/lib/certauth/.jwt_secret"
_ALGORITHM = "HS256"
_ACCESS_TOKEN_EXPIRE_MINUTES = 60
_ADMIN_USERNAME = "certauth"
_PKCS11_MODULE = "/usr/lib/aarch64-linux-gnu/opensc-pkcs11.so"
_YK_PUB_ROOT = "/tmp/yk1-root-pub.pem"
_YK_PUB_INT = "/tmp/yk2-int-pub.pem"
_TMP_DIR = "/var/lib/certauth/tmp"
def _get_jwt_secret() -> str:
"""Return JWT secret from env, persisted file, or generate new one."""
env_secret = os.environ.get("JWT_SECRET")
if env_secret:
return env_secret
if os.path.exists(_JWT_SECRET_FILE):
with open(_JWT_SECRET_FILE) as f:
return f.read().strip()
secret = secrets.token_hex(32)
os.makedirs(os.path.dirname(_JWT_SECRET_FILE), exist_ok=True)
fd = os.open(_JWT_SECRET_FILE, os.O_WRONLY | os.O_CREAT, 0o600)
with os.fdopen(fd, "w") as f:
f.write(secret)
return secret
def _get_pin(env_var: str) -> str:
"""Require YubiKey PIN from environment — no default allowed."""
pin = os.environ.get(env_var)
if not pin:
raise RuntimeError(f"Missing required environment variable: {env_var}")
return pin
YK_ROOT_SERIAL = _YK_ROOT_SERIAL
YK_INT_SERIAL = _YK_INT_SERIAL
ROOT_CA_PATH = _ROOT_CA_PATH
INT_CA_PATH = _INT_CA_PATH
CA_CHAIN_PATH = _CA_CHAIN_PATH
ISSUED_DIR = _ISSUED_DIR
DB_PATH = _DB_PATH
SECRET_KEY = _get_jwt_secret()
ALGORITHM = _ALGORITHM
ACCESS_TOKEN_EXPIRE_MINUTES = _ACCESS_TOKEN_EXPIRE_MINUTES
ADMIN_USERNAME = _ADMIN_USERNAME
PKCS11_MODULE = _PKCS11_MODULE
YK_PUB_ROOT = _YK_PUB_ROOT
YK_PUB_INT = _YK_PUB_INT
TMP_DIR = _TMP_DIR
YK_ROOT_PIN = _get_pin("YK_ROOT_PIN")
YK_INT_PIN = _get_pin("YK_INT_PIN")

View File

@ -1,4 +1,4 @@
import os, sqlite3, datetime, secrets, hashlib, subprocess, json
import os, sqlite3, datetime, secrets, hashlib, subprocess, json, time, functools
from fastapi import FastAPI, Request, Depends, HTTPException, Form
from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, JSONResponse, PlainTextResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
@ -14,12 +14,42 @@ from cryptography.hazmat.primitives import serialization
app = FastAPI(title="CertAuth Key Vault")
app.mount("/static", StaticFiles(directory="/opt/certauth/api/static"), name="static")
_login_attempts = {}
_LOGIN_MAX_ATTEMPTS = 5
_LOGIN_WINDOW_SECONDS = 900
_csrf_secret = secrets.token_hex(32)
def _check_rate_limit(client_ip: str) -> bool:
now = time.time()
if client_ip not in _login_attempts:
_login_attempts[client_ip] = []
_login_attempts[client_ip] = [
t for t in _login_attempts[client_ip] if now - t < _LOGIN_WINDOW_SECONDS
]
if len(_login_attempts[client_ip]) >= _LOGIN_MAX_ATTEMPTS:
return False
_login_attempts[client_ip].append(now)
return True
def _generate_csrf_token() -> str:
return secrets.token_hex(32)
def _verify_csrf_token(request: Request, token: str) -> bool:
stored = request.cookies.get("csrf_token")
if not stored or not token:
return False
return secrets.compare_digest(stored, token)
def _set_csrf_cookie(resp):
token = _generate_csrf_token()
resp.set_cookie("csrf_token", token, httponly=False, samesite="strict", path="/")
return token
def get_user_from_cookie(request: Request):
token = request.cookies.get("token")
if not token: return None
try: return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except: return None
jinja_env = Environment(
loader=FileSystemLoader("/opt/certauth/api/templates"),
@ -43,7 +73,10 @@ class LoginRequest(BaseModel):
password: str
@app.post("/api/token")
async def login(req: LoginRequest):
async def login(req: LoginRequest, request: Request):
client_ip = request.client.host
if not _check_rate_limit(client_ip):
raise HTTPException(429, "Too many login attempts. Try again later.")
conn = get_db()
row = conn.execute("SELECT * FROM users WHERE username = ?", (req.username,)).fetchone()
conn.close()
@ -103,7 +136,9 @@ async def sign_cert(cert_id: int, user: str = Depends(get_current_user)):
raise HTTPException(400, "Not found or already signed")
conn.close()
result, err = build_leaf_cert(row["subject"], row["san"], 365)
if err: raise HTTPException(500, f"Signing failed: {err}")
if err:
import html as h
raise HTTPException(500, f"Signing failed: {h.escape(str(err))}")
cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt"
kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
open(cf, "w").write(result["cert_pem"])
@ -203,7 +238,11 @@ async def login_page(request: Request):
return render("login.html", {"request": request, "error": None})
@app.post("/login")
async def login_post(username: str = Form(...), password: str = Form(...)):
async def login_post(username: str = Form(...), password: str = Form(...),
request: Request = None):
client_ip = request.client.host if request else "unknown"
if not _check_rate_limit(client_ip):
return render("login.html", {"request": None, "error": "Too many login attempts. Try again later."})
conn = get_db()
row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
conn.close()
@ -211,26 +250,31 @@ async def login_post(username: str = Form(...), password: str = Form(...)):
return render("login.html", {"request": None, "error": "Invalid credentials"})
token = create_access_token({"sub": username})
resp = RedirectResponse("/", status_code=302)
resp.set_cookie("token", token, httponly=True, samesite="lax", path="/")
resp.set_cookie("token", token, httponly=True, samesite="strict", secure=True, path="/")
_set_csrf_cookie(resp)
return resp
@app.post("/api/certs/{cert_id}/sign/web")
async def sign_cert_web(cert_id: int, request: Request = None):
async def sign_cert_web(cert_id: int, csrf_token: str = Form(""), request: Request = None):
user = get_user_from_cookie(request)
if not user:
return RedirectResponse("/login", status_code=302)
if not _verify_csrf_token(request, csrf_token):
raise HTTPException(403, "Invalid CSRF token")
conn = get_db()
row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
if not row or row["status"] != "pending":
conn.close()
raise HTTPException(400, "Not found or already issued")
conn.close()
import html as html_lib
try:
result, err = build_leaf_cert(row["subject"], row["san"], 365)
if err:
return HTMLResponse(f'<span class="text-red-400">Issue failed: {err}</span>')
safe_err = html_lib.escape(str(err))
return HTMLResponse(f'<span class="text-red-400">Issue failed: {safe_err}</span>')
cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt"
kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
open(cf, "w").write(result["cert_pem"])
@ -244,7 +288,8 @@ async def sign_cert_web(cert_id: int, request: Request = None):
conn2.close()
return HTMLResponse(f'<span class="text-green-400">Issued! <a href="/api/certs/{cert_id}/pem" class="underline">PEM</a> | <a href="/api/certs/{cert_id}/pfx" class="underline">PFX</a> | <a href="/certs" class="underline">Refresh</a></span>')
except Exception as ex:
return HTMLResponse(f'<span class="text-red-400">Issue failed: {str(ex)}</span>')
safe_ex = html_lib.escape(str(ex))
return HTMLResponse(f'<span class="text-red-400">Issue failed: {safe_ex}</span>')
@app.get("/logout")
@ -255,10 +300,13 @@ async def logout():
# --- Web API (cookie auth) ---
@app.post("/api/domains/web")
async def create_domain_web(name: str = Form(...), description: str = Form(""), request: Request = None):
async def create_domain_web(name: str = Form(...), description: str = Form(""),
csrf_token: str = Form(""), request: Request = None):
user = get_user_from_cookie(request)
if not user:
return RedirectResponse("/login", status_code=302)
if not _verify_csrf_token(request, csrf_token):
raise HTTPException(403, "Invalid CSRF token")
conn = get_db()
cur = conn.cursor()
cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", (name, description, 1))
@ -267,13 +315,16 @@ async def create_domain_web(name: str = Form(...), description: str = Form(""),
return HTMLResponse('<span class="text-green-400">Domain registered! <a href="/domains" class="underline">Refresh</a></span>')
@app.post("/api/certs/web/request")
async def request_cert_web(cn: str = Form(...), sans: str = Form(""), days: int = Form(365), domain_id: int = Form(0), request: Request = None):
async def request_cert_web(cn: str = Form(...), sans: str = Form(""), days: int = Form(365),
domain_id: int = Form(0), csrf_token: str = Form(""),
request: Request = None):
user = get_user_from_cookie(request)
if not user:
return RedirectResponse("/login", status_code=302)
if not _verify_csrf_token(request, csrf_token):
raise HTTPException(403, "Invalid CSRF token")
conn = get_db()
cur = conn.cursor()
# Look up domain by CN if domain_id not provided
if domain_id == 0:
cur.execute("SELECT id FROM domains WHERE name=?", (cn,))
row = cur.fetchone()

View File

@ -1,12 +1,11 @@
import subprocess, datetime, os, hashlib, ipaddress, re
import subprocess, datetime, os, hashlib, ipaddress, re, tempfile
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import NameOID
from config import *
TMP_DIR = "/var/lib/certauth/tmp"
os.makedirs(TMP_DIR, exist_ok=True)
def get_root_pub_key():
@ -26,24 +25,40 @@ def der_len(n):
elif n < 0x100: return bytes([0x81, n])
return bytes([0x82, n>>8, n&0xff])
def _make_temp_file(prefix: str, data: bytes = None):
"""Create temp file with unpredictable name in TMP_DIR."""
fd, path = tempfile.mkstemp(prefix=prefix, dir=TMP_DIR)
try:
if data is not None:
os.write(fd, data)
finally:
os.close(fd)
os.chmod(path, 0o600)
return path
def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
tbs_file = os.path.join(TMP_DIR, "tbs_sign.der")
sig_file = os.path.join(TMP_DIR, "sig_out.bin")
with open(tbs_file, "wb") as f:
f.write(tbs_bytes)
r = subprocess.run([
"sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
"--login", "--pin", yk_pin,
"--sign", "--mechanism", "ECDSA-SHA384",
"--token-label", token_label,
"--label", "SIGN key",
"--input-file", tbs_file,
"--output-file", sig_file
], capture_output=True, text=True)
if r.returncode != 0:
return None, r.stderr
with open(sig_file, "rb") as f:
raw = f.read()
tbs_file = _make_temp_file("tbs_", tbs_bytes)
sig_file = _make_temp_file("sig_")
os.unlink(sig_file)
try:
r = subprocess.run([
"sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
"--login", "--pin-source", "stdin",
"--sign", "--mechanism", "ECDSA-SHA384",
"--token-label", token_label,
"--label", "SIGN key",
"--input-file", tbs_file,
"--output-file", sig_file
], input=yk_pin, capture_output=True, text=True)
if r.returncode != 0:
return None, r.stderr
with open(sig_file, "rb") as f:
raw = f.read()
finally:
for f in (tbs_file, sig_file):
if os.path.exists(f):
os.unlink(f)
rb = raw[:48].lstrip(b"\x00") or b"\x00"
sb = raw[48:].lstrip(b"\x00") or b"\x00"
if rb[0] & 0x80: rb = b"\x00" + rb
@ -119,14 +134,19 @@ def build_leaf_cert(cn, sans, days=365):
content = tbs_full + alg_full + new_sig
cl = len(content)
final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content
der_file = os.path.join(TMP_DIR, "leaf.der")
pem_file = os.path.join(TMP_DIR, "leaf.pem")
with open(der_file, "wb") as f: f.write(final)
r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
"-in", der_file, "-out", pem_file],
capture_output=True, text=True)
if r.returncode != 0: return None, r.stderr
with open(pem_file) as f: leaf_pem = f.read()
der_file = _make_temp_file("leaf_", final)
pem_file = _make_temp_file("leaf_pem_")
os.unlink(pem_file)
try:
r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
"-in", der_file, "-out", pem_file],
capture_output=True, text=True)
if r.returncode != 0: return None, r.stderr
with open(pem_file) as f: leaf_pem = f.read()
finally:
for f in (der_file, pem_file):
if os.path.exists(f):
os.unlink(f)
key_pem = leaf_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,

View File

@ -3,7 +3,7 @@
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Playground</title>
<title>Playground Network</title>
<style>
:root { --bg: #0f172a; --card: #1e293b; --text: #f8fafc; --accent: #38bdf8; }
body { font-family: system-ui, sans-serif; background: var(--bg); color: var(--text); margin: 0; padding: 2rem; }
@ -19,15 +19,13 @@
</head>
<body>
<div class="container">
<h1>Playground</h1>
<h1>Playground Network</h1>
<div class="grid">
<a href="https://pinvault.example.com" class="card"><div class="icon">🔐</div><h2>PinVault</h2><p>Password manager</p></a>
<a href="https://youtube.example.com" class="card"><div class="icon">📺</div><h2>YouTube</h2><p>YouTube CLI</p></a>
<a href="https://archive.example.com" class="card"><div class="icon">🏛️</div><h2>Archive</h2><p>Website archiving</p></a>
<a href="https://vote.example.com" class="card"><div class="icon">🗳️</div><h2>Vote</h2><p>Voting app</p></a>
<a href="https://search.example.com" class="card"><div class="icon">🔍</div><h2>Search</h2><p>Privacy metasearch</p></a>
<a href="https://nextcloud.example.com" class="card"><div class="icon">☁️</div><h2>Nextcloud</h2><p>Cloud storage</p></a>
<a href="https://opencloud.example.com" class="card"><div class="icon">🌐</div><h2>OpenCloud</h2><p>OpenCloud service</p></a>
<a href="https://paste.example.com" class="card"><div class="icon">📋</div><h2>Paste</h2><p>Pastebin service</p></a>
<a href="https://ai.example.com" class="card"><div class="icon">🤖</div><h2>AI</h2><p>Local LLM</p></a>
<a href="https://archive.example.com" class="card"><div class="icon">🏛️</div><h2>Archive</h2><p>Website archiving</p></a>
</div>
</div>
</body>

View File

@ -0,0 +1,5 @@
-----BEGIN PRIVATE KEY-----
MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQggL95oQbMsPLm3Kd0
ihyuirXpBdKFE2IAqZyVZwqtAWKhRANCAAR+r1V3/0xyeoa50fq+XdZ90U3di7wQ
q2mQVmwMFTCNBrNkOjg6AruiPP1aURCKbT3C+TlRFTFE7iLKFCkGeihp
-----END PRIVATE KEY-----

12
landing/playground.ms.pem Normal file
View File

@ -0,0 +1,12 @@
-----BEGIN CERTIFICATE-----
MIIBsjCCAVmgAwIBAgIUJpVaTpjyW1GLONCx3rXwKnc4OwAwCgYIKoZIzj0EAwIw
GDEWMBQGA1UEAwwNcGxheWdyb3VuZC5tczAeFw0yNjA3MDIxOTQyMThaFw0yNzA3
MDIxOTQyMThaMBgxFjAUBgNVBAMMDXBsYXlncm91bmQubXMwWTATBgcqhkjOPQIB
BggqhkjOPQMBBwNCAAR+r1V3/0xyeoa50fq+XdZ90U3di7wQq2mQVmwMFTCNBrNk
Ojg6AruiPP1aURCKbT3C+TlRFTFE7iLKFCkGeihpo4GAMH4wHQYDVR0OBBYEFA19
0spDELs37Zq3rci0GxkppbGnMB8GA1UdIwQYMBaAFA190spDELs37Zq3rci0Gxkp
pbGnMA8GA1UdEwEB/wQFMAMBAf8wKwYDVR0RBCQwIoINcGxheWdyb3VuZC5tc4IR
d3d3LnBsYXlncm91bmQubXMwCgYIKoZIzj0EAwIDRwAwRAIgfSL/SDyLeCmg4l3S
8U9KvBEQkJe1yaId8QmFhhUV+gICIE5KFmPJ3GrEbjzJyB1DFjHTDNUocJBgut0w
09Y9Lz5F
-----END CERTIFICATE-----

View File

@ -0,0 +1,32 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Playground Network</title>
<style>
:root { --bg: #0f172a; --card: #1e293b; --text: #f8fafc; --accent: #38bdf8; }
body { font-family: system-ui, sans-serif; background: var(--bg); color: var(--text); margin: 0; padding: 2rem; }
.container { max-width: 1200px; margin: 0 auto; }
h1 { text-align: center; margin-bottom: 3rem; font-size: 2.5rem; background: linear-gradient(135deg, var(--accent), #818cf8); -webkit-background-clip: text; -webkit-text-fill-color: transparent; }
.grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); gap: 1.5rem; }
.card { background: var(--card); border-radius: 12px; padding: 1.5rem; text-decoration: none; color: var(--text); transition: all 0.2s; border: 1px solid rgba(255,255,255,0.1); }
.card:hover { transform: translateY(-2px); border-color: var(--accent); box-shadow: 0 4px 12px rgba(56, 189, 248, 0.1); }
.card h2 { margin: 0 0 0.5rem; font-size: 1.25rem; }
.card p { margin: 0; color: #94a3b8; font-size: 0.9rem; }
.icon { font-size: 2rem; margin-bottom: 1rem; }
</style>
</head>
<body>
<div class="container">
<h1>Playground Network</h1>
<div class="grid">
<a href="https://pinvault.example.com" class="card"><div class="icon">🔐</div><h2>PinVault</h2><p>Password manager</p></a>
<a href="https://nextcloud.example.com" class="card"><div class="icon">☁️</div><h2>Nextcloud</h2><p>Cloud storage</p></a>
<a href="https://opencloud.example.com" class="card"><div class="icon">🌐</div><h2>OpenCloud</h2><p>OpenCloud service</p></a>
<a href="https://paste.example.com" class="card"><div class="icon">📋</div><h2>Paste</h2><p>Pastebin service</p></a>
<a href="https://archive.example.com" class="card"><div class="icon">🏛️</div><h2>Archive</h2><p>Website archiving</p></a>
</div>
</div>
</body>
</html>

26
nginx-playground.ms Normal file
View File

@ -0,0 +1,26 @@
# ============================================================
# example.com - Main landing page (HTTPS)
# ============================================================
server {
listen 443 ssl;
server_name example.com www.example.com;
ssl_certificate /etc/ssl/certs/example.com.pem;
ssl_certificate_key /etc/ssl/private/example.com.key;
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers HIGH:!aNULL:!MD5;
root /var/www/example.com;
index index.html index.htm;
location / {
try_files $uri $uri/ =404;
}
}
# HTTP -> HTTPS redirect for example.com
server {
listen 80;
server_name example.com www.example.com;
return 301 https://$host$request_uri;
}

59
tests/test_auth.py Normal file
View File

@ -0,0 +1,59 @@
import os
import sys
import unittest
from unittest import mock
from httpx import AsyncClient, ASGITransport
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "api"))
class TestAuthSecurity(unittest.TestCase):
def test_rate_limiting_exists(self):
import main
self.assertTrue(hasattr(main, '_check_rate_limit'))
self.assertEqual(main._LOGIN_MAX_ATTEMPTS, 5)
self.assertEqual(main._LOGIN_WINDOW_SECONDS, 900)
def test_csrf_token_generation(self):
import main
t1 = main._generate_csrf_token()
t2 = main._generate_csrf_token()
self.assertNotEqual(t1, t2)
self.assertEqual(len(t1), 64)
def test_csrf_verify_rejects_empty(self):
import main
req = mock.MagicMock()
req.cookies.get.return_value = None
self.assertFalse(main._verify_csrf_token(req, "any-token"))
def test_csrf_verify_rejects_mismatch(self):
import main
req = mock.MagicMock()
req.cookies.get.return_value = "stored-token"
self.assertFalse(main._verify_csrf_token(req, "different-token"))
def test_csrf_verify_accepts_match(self):
import main
req = mock.MagicMock()
req.cookies.get.return_value = "matching-token"
self.assertTrue(main._verify_csrf_token(req, "matching-token"))
def test_no_duplicate_get_user_from_cookie(self):
import main, inspect
sources = inspect.getsourcelines(main)[0]
count = sum(1 for line in sources if 'def get_user_from_cookie' in line)
self.assertEqual(count, 1, "get_user_from_cookie must be defined exactly once")
def test_xss_escaped_in_sign_error(self):
"""Check that HTML in error messages gets escaped."""
import html as h
err = '<script>alert("xss")</script>'
escaped = h.escape(err)
self.assertNotIn("<script>", escaped)
self.assertIn("&lt;script&gt;", escaped)
if __name__ == "__main__":
unittest.main()

66
tests/test_config.py Normal file
View File

@ -0,0 +1,66 @@
import os
import sys
import tempfile
import unittest
class TestConfigSecurity(unittest.TestCase):
def test_jwt_secret_generates_new(self):
with tempfile.NamedTemporaryFile(suffix=".jwt_secret", delete=False) as f:
secret_path = f.name
os.unlink(secret_path)
try:
os.environ.pop("JWT_SECRET", None)
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import importlib
if "config" in sys.modules:
del sys.modules["config"]
os.environ["_JWT_SECRET_FILE"] = secret_path
cfg = __import__("config")
self.assertIsNotNone(cfg.SECRET_KEY)
self.assertEqual(len(cfg.SECRET_KEY), 64)
with open(secret_path) as sf:
self.assertEqual(sf.read().strip(), cfg.SECRET_KEY)
finally:
os.environ.pop("JWT_SECRET", None)
os.environ.pop("_JWT_SECRET_FILE", None)
if os.path.exists(secret_path):
os.unlink(secret_path)
if "config" in sys.modules:
del sys.modules["config"]
def test_jwt_secret_env_override(self):
os.environ["JWT_SECRET"] = "test-secret-from-env"
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
if "config" in sys.modules:
del sys.modules["config"]
cfg = __import__("config")
self.assertEqual(cfg.SECRET_KEY, "test-secret-from-env")
os.environ.pop("JWT_SECRET", None)
if "config" in sys.modules:
del sys.modules["config"]
def test_yk_pin_requires_env(self):
os.environ.pop("YK_ROOT_PIN", None)
os.environ.pop("YK_INT_PIN", None)
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
if "config" in sys.modules:
del sys.modules["config"]
with self.assertRaises(RuntimeError):
__import__("config")
os.environ["YK_ROOT_PIN"] = "test1234"
os.environ["YK_INT_PIN"] = "test5678"
if "config" in sys.modules:
del sys.modules["config"]
cfg = __import__("config")
self.assertEqual(cfg.YK_ROOT_PIN, "test1234")
self.assertEqual(cfg.YK_INT_PIN, "test5678")
os.environ.pop("YK_ROOT_PIN", None)
os.environ.pop("YK_INT_PIN", None)
if "config" in sys.modules:
del sys.modules["config"]
if __name__ == "__main__":
unittest.main()

51
tests/test_signing.py Normal file
View File

@ -0,0 +1,51 @@
import os
import sys
import unittest
from unittest import mock
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "api"))
class TestSigningSecurity(unittest.TestCase):
@mock.patch("signing.subprocess.run")
@mock.patch("signing._make_temp_file")
def test_pin_not_in_cli_args(self, mock_mkstemp, mock_run):
"""PIN must NOT appear in subprocess command-line args."""
mock_mkstemp.return_value = "/tmp/test_file"
mock_run.return_value = mock.MagicMock(returncode=0)
import signing
signing.sign_tbs_with_yk(b"\x00" * 100, "test-pin-123")
cmd = mock_run.call_args[0][0]
self.assertNotIn("test-pin-123", cmd, "PIN must not appear in command args")
self.assertIn("--pin-source", cmd)
self.assertIn("stdin", cmd)
self.assertNotIn("--pin", cmd) or cmd.index("--pin-source") < cmd.index("--pin")
@mock.patch("signing.subprocess.run")
def test_temp_files_use_mkstemp(self, mock_run):
"""Temp files must use mkstemp, not predictable names."""
mock_run.return_value = mock.MagicMock(returncode=0)
with mock.patch("signing.tempfile.mkstemp") as mock_mkstemp:
mock_mkstemp.return_value = (0, "/tmp/unpredictable_name")
import signing
try:
signing.sign_tbs_with_yk(b"\x00" * 100, "pin")
except Exception:
pass
calls = [c[0][1] for c in mock_mkstemp.call_args_list]
self.assertTrue(all("/tmp/unpredictable_name" in c for c in calls),
"All temp files should use mkstemp")
def test_no_hardcoded_pin_default(self):
"""YK PIN must fail if env var not set."""
os.environ.pop("YK_ROOT_PIN", None)
os.environ.pop("YK_INT_PIN", None)
if "config" in sys.modules:
del sys.modules["config"]
with self.assertRaises(RuntimeError):
__import__("config")
if __name__ == "__main__":
unittest.main()