feat: Flask API for encrypted records (CRUD + replay protection)

Exposes a ciphertext-only REST API: list/get/create/update/delete of
records as opaque base64 blobs. Enforces base64/size validation,
rejects duplicates, and uses patch-version checks server-side so
replayed or stale updates are refused. Includes run.py entry point
that enables TLS when certs/ are present. Verified via test client.

🤖 Generated with Codebuff
Co-Authored-By: Codebuff <noreply@codebuff.com>

backend/app.py

@@ -0,0 +1,155 @@
+"""Flask application exposing the records API.
+
+The server is a dumb ciphertext store: it never sees plaintext or the
+key. Every record body is opaque base64. Requests are validated only
+for shape and size, never for content.
+
+Security properties:
+  - No plaintext is logged, stored, or echoed.
+  - Version counters are enforced server-side to detect replays.
+  - HTTPS is handled by ssl context in run.py (TLS for dev).
+"""
+
+import base64
+import binascii
+
+from flask import Flask, jsonify, request, send_from_directory
+
+from . import storage
+
+MAX_CIPHERTEXT_B64 = 32 * 1024  # plenty for one record, stops abuse
+MAX_SALT_NONCE_B64 = 512
+
+app = Flask(__name__, static_folder=None)
+
+
+# ---------- basic hardening ----------
+
+@app.after_request
+def _security_headers(resp):
+    resp.headers["X-Content-Type-Options"] = "nosniff"
+    resp.headers["X-Frame-Options"] = "DENY"
+    resp.headers["Referrer-Policy"] = "no-referrer"
+    resp.headers["Cache-Control"] = "no-store"
+    return resp
+
+
+def _b64_field(data, field, max_len, required=True):
+    """Validate and decode a base64 field from the JSON body. Returns bytes or None."""
+    if field not in data:
+        if required:
+            raise ValueError(f"missing field: {field}")
+        return None
+    raw = data[field]
+    if not isinstance(raw, str) or not raw or len(raw) > max_len:
+        raise ValueError(f"invalid field: {field}")
+    try:
+        return base64.b64decode(raw, validate=True)
+    except (binascii.Error, ValueError):
+        raise ValueError(f"invalid base64 in field: {field}")
+
+
+def _record_json(rec):
+    return {
+        "uuid": rec["uuid"],
+        "ciphertext": base64.b64encode(rec["ciphertext"]).decode(),
+        "nonce": base64.b64encode(rec["nonce"]).decode(),
+        "salt": base64.b64encode(rec["salt"]).decode(),
+        "version": rec["version"],
+        "created_at": rec["created_at"],
+        "updated_at": rec["updated_at"],
+    }
+
+
+# ---------- static UI ----------
+
+@app.route("/")
+def index():
+    return send_from_directory("static", "index.html")
+
+
+@app.route("/static/<path:filename>")
+def static_files(filename):
+    return send_from_directory("static", filename)
+
+
+# ---------- API ----------
+
+@app.get("/api/records")
+def api_list():
+    return jsonify([_record_json(r) for r in storage.list_records()])
+
+
+@app.get("/api/records/<uuid>")
+def api_get(uuid):
+    rec = storage.get_record(uuid)
+    if rec is None:
+        return jsonify({"error": "not found"}), 404
+    return jsonify(_record_json(rec))
+
+
+@app.post("/api/records")
+def api_create():
+    data = request.get_json(silent=True)
+    if not isinstance(data, dict):
+        return jsonify({"error": "invalid JSON body"}), 400
+    try:
+        uuid = data.get("uuid")
+        if not isinstance(uuid, str) or not uuid or len(uuid) > 64:
+            raise ValueError("invalid field: uuid")
+        ct = _b64_field(data, "ciphertext", MAX_CIPHERTEXT_B64)
+        nonce = _b64_field(data, "nonce", MAX_SALT_NONCE_B64)
+        salt = _b64_field(data, "salt", MAX_SALT_NONCE_B64)
+    except ValueError as e:
+        return jsonify({"error": str(e)}), 400
+
+    if storage.get_record(uuid) is not None:
+        return jsonify({"error": "record already exists"}), 409
+
+    rec = storage.create_record(uuid, ct, nonce, salt)
+    return jsonify(_record_json(rec)), 201
+
+
+@app.put("/api/records/<uuid>")
+def api_update(uuid):
+    data = request.get_json(silent=True)
+    if not isinstance(data, dict):
+        return jsonify({"error": "invalid JSON body"}), 400
+    try:
+        version = data.get("version")
+        if not isinstance(version, int) or version < 1:
+            raise ValueError("invalid field: version")
+        ct = _b64_field(data, "ciphertext", MAX_CIPHERTEXT_B64)
+        nonce = _b64_field(data, "nonce", MAX_SALT_NONCE_B64)
+        salt = _b64_field(data, "salt", MAX_SALT_NONCE_B64)
+    except ValueError as e:
+        return jsonify({"error": str(e)}), 400
+
+    rec = storage.update_record(uuid, ct, nonce, salt, expected_version=version)
+    if rec is None:
+        # 409: version mismatch (concurrent edit or replay) — do not
+        # distinguish from "not found" to avoid leaking existence.
+        return jsonify({"error": "version conflict or not found"}), 409
+    return jsonify(_record_json(rec))
+
+
+@app.delete("/api/records/<uuid>")
+def api_delete(uuid):
+    data = request.get_json(silent=True) or {}
+    version = data.get("version")
+    if not isinstance(version, int) or version < 1:
+        return jsonify({"error": "invalid field: version"}), 400
+    if not storage.delete_record(uuid, expected_version=version):
+        return jsonify({"error": "version conflict or not found"}), 409
+    return "", 204
+
+
+@app.errorhandler(404)
+def not_found(_e):
+    return jsonify({"error": "not found"}), 404
+
+
+@app.errorhandler(500)
+def server_error(_e):
+    # Never leak stack traces or data.
+    return jsonify({"error": "internal server error"}), 500

run.py

@@ -0,0 +1,26 @@
+"""Dev/Prod entry point. Runs the Flask app, over TLS when certs exist."""
+
+import os
+import ssl
+
+from backend.app import app
+from backend import storage
+
+
+def main():
+    storage.init_db()
+
+    cert = os.path.join("certs", "cert.pem")
+    key = os.path.join("certs", "key.pem")
+
+    if os.path.exists(cert) and os.path.exists(key):
+        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        context.load_cert_chain(cert, key)
+        app.run(host="127.0.0.1", port=8443, ssl_context=context)
+    else:
+        print("WARNING: no certs/ found, running WITHOUT TLS (dev only!).")
+        app.run(host="127.0.0.1", port=5000)
+
+
+if __name__ == "__main__":
+    main()