Compare commits
4 Commits
96f1102fb1
...
df82068aeb
| Author | SHA1 | Date | |
|---|---|---|---|
| df82068aeb | |||
| c759597ad0 | |||
| 279a28515f | |||
| c4d952430e |
15
Caddyfile-playground.ms
Normal file
15
Caddyfile-playground.ms
Normal 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
|
||||||
|
}
|
||||||
@ -1,18 +1,63 @@
|
|||||||
import os
|
import os
|
||||||
|
import secrets
|
||||||
|
|
||||||
YK_ROOT_SERIAL = "35450561"
|
_YK_ROOT_SERIAL = "35450561"
|
||||||
YK_ROOT_PIN = os.environ.get("YK_ROOT_PIN", "CHANGE_ME_YK1_PIN")
|
_YK_INT_SERIAL = "33930436"
|
||||||
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"
|
||||||
ROOT_CA_PATH = "/etc/ssl/ca/root/root-ca.crt"
|
_INT_CA_PATH = "/etc/ssl/ca/intermediate/intermediate-ca.crt"
|
||||||
INT_CA_PATH = "/etc/ssl/ca/intermediate/intermediate-ca.crt"
|
_CA_CHAIN_PATH = "/etc/ssl/ca/ca-chain.crt"
|
||||||
CA_CHAIN_PATH = "/etc/ssl/ca/ca-chain.crt"
|
_ISSUED_DIR = "/etc/ssl/ca/issued"
|
||||||
ISSUED_DIR = "/etc/ssl/ca/issued"
|
_DB_PATH = "/var/lib/certauth/certauth.db"
|
||||||
DB_PATH = "/var/lib/certauth/certauth.db"
|
_JWT_SECRET_FILE = "/var/lib/certauth/.jwt_secret"
|
||||||
SECRET_KEY = os.environ.get("JWT_SECRET", "CHANGE_ME_JWT_SECRET")
|
_ALGORITHM = "HS256"
|
||||||
ALGORITHM = "HS256"
|
_ACCESS_TOKEN_EXPIRE_MINUTES = 60
|
||||||
ACCESS_TOKEN_EXPIRE_MINUTES = 60
|
_ADMIN_USERNAME = "certauth"
|
||||||
ADMIN_USERNAME = "certauth"
|
_PKCS11_MODULE = "/usr/lib/aarch64-linux-gnu/opensc-pkcs11.so"
|
||||||
PKCS11_MODULE = "/usr/lib/aarch64-linux-gnu/opensc-pkcs11.so"
|
_YK_PUB_ROOT = "/tmp/yk1-root-pub.pem"
|
||||||
YK_PUB_ROOT = "/tmp/yk1-root-pub.pem"
|
_YK_PUB_INT = "/tmp/yk2-int-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")
|
||||||
|
|||||||
83
api/main.py
83
api/main.py
@ -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 import FastAPI, Request, Depends, HTTPException, Form
|
||||||
from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, JSONResponse, PlainTextResponse, StreamingResponse
|
from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, JSONResponse, PlainTextResponse, StreamingResponse
|
||||||
from fastapi.staticfiles import StaticFiles
|
from fastapi.staticfiles import StaticFiles
|
||||||
@ -14,12 +14,42 @@ from cryptography.hazmat.primitives import serialization
|
|||||||
app = FastAPI(title="CertAuth Key Vault")
|
app = FastAPI(title="CertAuth Key Vault")
|
||||||
app.mount("/static", StaticFiles(directory="/opt/certauth/api/static"), name="static")
|
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(
|
jinja_env = Environment(
|
||||||
loader=FileSystemLoader("/opt/certauth/api/templates"),
|
loader=FileSystemLoader("/opt/certauth/api/templates"),
|
||||||
@ -43,7 +73,10 @@ class LoginRequest(BaseModel):
|
|||||||
password: str
|
password: str
|
||||||
|
|
||||||
@app.post("/api/token")
|
@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()
|
conn = get_db()
|
||||||
row = conn.execute("SELECT * FROM users WHERE username = ?", (req.username,)).fetchone()
|
row = conn.execute("SELECT * FROM users WHERE username = ?", (req.username,)).fetchone()
|
||||||
conn.close()
|
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")
|
raise HTTPException(400, "Not found or already signed")
|
||||||
conn.close()
|
conn.close()
|
||||||
result, err = build_leaf_cert(row["subject"], row["san"], 365)
|
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"
|
cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt"
|
||||||
kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
|
kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
|
||||||
open(cf, "w").write(result["cert_pem"])
|
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})
|
return render("login.html", {"request": request, "error": None})
|
||||||
|
|
||||||
@app.post("/login")
|
@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()
|
conn = get_db()
|
||||||
row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
|
row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
|
||||||
conn.close()
|
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"})
|
return render("login.html", {"request": None, "error": "Invalid credentials"})
|
||||||
token = create_access_token({"sub": username})
|
token = create_access_token({"sub": username})
|
||||||
resp = RedirectResponse("/", status_code=302)
|
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
|
return resp
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/certs/{cert_id}/sign/web")
|
@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)
|
user = get_user_from_cookie(request)
|
||||||
if not user:
|
if not user:
|
||||||
return RedirectResponse("/login", status_code=302)
|
return RedirectResponse("/login", status_code=302)
|
||||||
|
if not _verify_csrf_token(request, csrf_token):
|
||||||
|
raise HTTPException(403, "Invalid CSRF token")
|
||||||
conn = get_db()
|
conn = get_db()
|
||||||
row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
|
row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
|
||||||
if not row or row["status"] != "pending":
|
if not row or row["status"] != "pending":
|
||||||
conn.close()
|
conn.close()
|
||||||
raise HTTPException(400, "Not found or already issued")
|
raise HTTPException(400, "Not found or already issued")
|
||||||
conn.close()
|
conn.close()
|
||||||
|
import html as html_lib
|
||||||
try:
|
try:
|
||||||
result, err = build_leaf_cert(row["subject"], row["san"], 365)
|
result, err = build_leaf_cert(row["subject"], row["san"], 365)
|
||||||
if err:
|
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"
|
cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt"
|
||||||
kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
|
kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
|
||||||
open(cf, "w").write(result["cert_pem"])
|
open(cf, "w").write(result["cert_pem"])
|
||||||
@ -244,7 +288,8 @@ async def sign_cert_web(cert_id: int, request: Request = None):
|
|||||||
conn2.close()
|
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>')
|
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:
|
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")
|
@app.get("/logout")
|
||||||
@ -255,10 +300,13 @@ async def logout():
|
|||||||
|
|
||||||
# --- Web API (cookie auth) ---
|
# --- Web API (cookie auth) ---
|
||||||
@app.post("/api/domains/web")
|
@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)
|
user = get_user_from_cookie(request)
|
||||||
if not user:
|
if not user:
|
||||||
return RedirectResponse("/login", status_code=302)
|
return RedirectResponse("/login", status_code=302)
|
||||||
|
if not _verify_csrf_token(request, csrf_token):
|
||||||
|
raise HTTPException(403, "Invalid CSRF token")
|
||||||
conn = get_db()
|
conn = get_db()
|
||||||
cur = conn.cursor()
|
cur = conn.cursor()
|
||||||
cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", (name, description, 1))
|
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>')
|
return HTMLResponse('<span class="text-green-400">Domain registered! <a href="/domains" class="underline">Refresh</a></span>')
|
||||||
|
|
||||||
@app.post("/api/certs/web/request")
|
@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)
|
user = get_user_from_cookie(request)
|
||||||
if not user:
|
if not user:
|
||||||
return RedirectResponse("/login", status_code=302)
|
return RedirectResponse("/login", status_code=302)
|
||||||
|
if not _verify_csrf_token(request, csrf_token):
|
||||||
|
raise HTTPException(403, "Invalid CSRF token")
|
||||||
conn = get_db()
|
conn = get_db()
|
||||||
cur = conn.cursor()
|
cur = conn.cursor()
|
||||||
# Look up domain by CN if domain_id not provided
|
|
||||||
if domain_id == 0:
|
if domain_id == 0:
|
||||||
cur.execute("SELECT id FROM domains WHERE name=?", (cn,))
|
cur.execute("SELECT id FROM domains WHERE name=?", (cn,))
|
||||||
row = cur.fetchone()
|
row = cur.fetchone()
|
||||||
|
|||||||
@ -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 import x509
|
||||||
from cryptography.hazmat.primitives import hashes, serialization
|
from cryptography.hazmat.primitives import hashes, serialization
|
||||||
from cryptography.hazmat.primitives.asymmetric import ec
|
from cryptography.hazmat.primitives.asymmetric import ec
|
||||||
from cryptography.x509.oid import NameOID
|
from cryptography.x509.oid import NameOID
|
||||||
from config import *
|
from config import *
|
||||||
|
|
||||||
TMP_DIR = "/var/lib/certauth/tmp"
|
|
||||||
os.makedirs(TMP_DIR, exist_ok=True)
|
os.makedirs(TMP_DIR, exist_ok=True)
|
||||||
|
|
||||||
def get_root_pub_key():
|
def get_root_pub_key():
|
||||||
@ -26,24 +25,40 @@ def der_len(n):
|
|||||||
elif n < 0x100: return bytes([0x81, n])
|
elif n < 0x100: return bytes([0x81, n])
|
||||||
return bytes([0x82, n>>8, n&0xff])
|
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"):
|
def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
|
||||||
tbs_file = os.path.join(TMP_DIR, "tbs_sign.der")
|
tbs_file = _make_temp_file("tbs_", tbs_bytes)
|
||||||
sig_file = os.path.join(TMP_DIR, "sig_out.bin")
|
sig_file = _make_temp_file("sig_")
|
||||||
with open(tbs_file, "wb") as f:
|
os.unlink(sig_file)
|
||||||
f.write(tbs_bytes)
|
try:
|
||||||
r = subprocess.run([
|
r = subprocess.run([
|
||||||
"sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
|
"sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
|
||||||
"--login", "--pin", yk_pin,
|
"--login", "--pin-source", "stdin",
|
||||||
"--sign", "--mechanism", "ECDSA-SHA384",
|
"--sign", "--mechanism", "ECDSA-SHA384",
|
||||||
"--token-label", token_label,
|
"--token-label", token_label,
|
||||||
"--label", "SIGN key",
|
"--label", "SIGN key",
|
||||||
"--input-file", tbs_file,
|
"--input-file", tbs_file,
|
||||||
"--output-file", sig_file
|
"--output-file", sig_file
|
||||||
], capture_output=True, text=True)
|
], input=yk_pin, capture_output=True, text=True)
|
||||||
if r.returncode != 0:
|
if r.returncode != 0:
|
||||||
return None, r.stderr
|
return None, r.stderr
|
||||||
with open(sig_file, "rb") as f:
|
with open(sig_file, "rb") as f:
|
||||||
raw = f.read()
|
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"
|
rb = raw[:48].lstrip(b"\x00") or b"\x00"
|
||||||
sb = 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
|
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
|
content = tbs_full + alg_full + new_sig
|
||||||
cl = len(content)
|
cl = len(content)
|
||||||
final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content
|
final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content
|
||||||
der_file = os.path.join(TMP_DIR, "leaf.der")
|
der_file = _make_temp_file("leaf_", final)
|
||||||
pem_file = os.path.join(TMP_DIR, "leaf.pem")
|
pem_file = _make_temp_file("leaf_pem_")
|
||||||
with open(der_file, "wb") as f: f.write(final)
|
os.unlink(pem_file)
|
||||||
r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
|
try:
|
||||||
"-in", der_file, "-out", pem_file],
|
r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
|
||||||
capture_output=True, text=True)
|
"-in", der_file, "-out", pem_file],
|
||||||
if r.returncode != 0: return None, r.stderr
|
capture_output=True, text=True)
|
||||||
with open(pem_file) as f: leaf_pem = f.read()
|
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(
|
key_pem = leaf_key.private_bytes(
|
||||||
encoding=serialization.Encoding.PEM,
|
encoding=serialization.Encoding.PEM,
|
||||||
format=serialization.PrivateFormat.PKCS8,
|
format=serialization.PrivateFormat.PKCS8,
|
||||||
|
|||||||
@ -3,7 +3,7 @@
|
|||||||
<head>
|
<head>
|
||||||
<meta charset="UTF-8">
|
<meta charset="UTF-8">
|
||||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||||
<title>Playground</title>
|
<title>Playground Network</title>
|
||||||
<style>
|
<style>
|
||||||
:root { --bg: #0f172a; --card: #1e293b; --text: #f8fafc; --accent: #38bdf8; }
|
: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; }
|
body { font-family: system-ui, sans-serif; background: var(--bg); color: var(--text); margin: 0; padding: 2rem; }
|
||||||
@ -19,15 +19,13 @@
|
|||||||
</head>
|
</head>
|
||||||
<body>
|
<body>
|
||||||
<div class="container">
|
<div class="container">
|
||||||
<h1>Playground</h1>
|
<h1>Playground Network</h1>
|
||||||
<div class="grid">
|
<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://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://nextcloud.example.com" class="card"><div class="icon">☁️</div><h2>Nextcloud</h2><p>Cloud storage</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://opencloud.example.com" class="card"><div class="icon">🌐</div><h2>OpenCloud</h2><p>OpenCloud service</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://paste.example.com" class="card"><div class="icon">📋</div><h2>Paste</h2><p>Pastebin 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>
|
||||||
</div>
|
</div>
|
||||||
</body>
|
</body>
|
||||||
|
|||||||
5
landing/playground.ms.key
Normal file
5
landing/playground.ms.key
Normal 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
12
landing/playground.ms.pem
Normal 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-----
|
||||||
32
landing/www/playground.ms/index.html
Normal file
32
landing/www/playground.ms/index.html
Normal 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
26
nginx-playground.ms
Normal 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
59
tests/test_auth.py
Normal 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("<script>", escaped)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
66
tests/test_config.py
Normal file
66
tests/test_config.py
Normal 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
51
tests/test_signing.py
Normal 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()
|
||||||
Loading…
x
Reference in New Issue
Block a user