Skip to content

Developer Patterns

Jump to bottom
NeySlim edited this page May 3, 2026 · 1 revision

Developer Patterns

Reusable backend helpers and conventions every contributor must follow when touching backend/api/v2/*, backend/services/*, or any code that handles credentials, encryption keys, or database transactions.

This page documents the canonical helpers introduced in v2.142 / v2.143 / v2.144 and explains what each helper exists to prevent, so reviewers can spot regressions in PRs.

Table of Contents

Why these helpers exist

Several production incidents in the v2.140 → v2.143 window were caused by the same code pattern being inlined in many places with subtle drift between sites:

Issue Pattern that drifted Helper introduced
#103, #104 Migration runner connection contract on PostgreSQL (fixed inline; tests added)
#105 Mixed encrypt() (base64-input) vs encrypt_string() (text-input) on PEM blobs encrypt_text() / decrypt_text()
latent base64.b64decode(decrypt_private_key(model.prv)) repeated 26× with no error context utils.key_codec.load_pem_bytes()
latent Bare db.session.commit() in service layer with no rollback on IntegrityError utils.db_transaction.commit_or_rollback()
latent Bare db.session.commit() in API handlers utils.safe_commit.safe_commit()
#106-class Silent except Exception: pass in auth/CSRF/email/syslog made post-mortems impossible logger.warning(..., exc_info=True) everywhere

Rule of thumb: if you're about to write base64.b64decode(decrypt_private_key(...)) or db.session.commit() or except Exception: pass, stop. There is a helper. Use it.

PEM key load/store — utils.key_codec

Problem this prevents

Before v2.144, every site that had to materialise a stored private key did this:

import base64
from security.encryption import decrypt_private_key
pem = base64.b64decode(decrypt_private_key(ca.prv))
key = serialization.load_pem_private_key(pem, password=None, backend=default_backend())

Drift symptoms in production:

  • When ca.prv is malformed (corruption, manual DB edit, restore from a different KEY_ENCRYPTION_KEY), the inline pattern raised binascii.Error: Invalid base64-encoded string with no indication of which CA / certificate.
  • Some sites wrapped it in try/except, others didn't. Some logged the model id, others the model object's __repr__.
  • 26 import sites for base64 + security.encryption that were really doing one thing.

How to use

from utils.key_codec import load_pem_bytes, store_pem_bytes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization

# Load — always pass a context string for diagnostic errors.
pem = load_pem_bytes(ca.prv, context=f"CA {ca.id}")
key = serialization.load_pem_private_key(pem, password=None, backend=default_backend())

# Store — symmetric inverse.
ca.prv = store_pem_bytes(pem_bytes)

load_pem_bytes():

  • Accepts the wire format (base64(encrypt_private_key(pem))) used by CA.prv and Certificate.prv.
  • Raises ValueError with the supplied context ("CA 42", "certificate 17", "SCEP CA 3") — never an opaque binascii.Error.
  • Tolerates plaintext PEM round-trips through the v2.143 passthrough fix in decrypt_private_key().

When NOT to use

Test coverage

backend/tests/test_key_codec.py — 8 tests including a TestEquivalenceWithLegacyPattern class that asserts byte-for-byte parity with the inline pattern, both with and without KEY_ENCRYPTION_KEY set.

Database commits — utils.db_transaction

Problem this prevents

A bare db.session.commit() that hits a constraint violation leaves the SQLAlchemy session in an aborted state. The next query in the same request raises InvalidRequestError: This Session's transaction has been rolled back due to a previous exception. Before v2.144:

  • 10 sites in auth/unified.py, services/mtls_auth_service.py, services/webauthn_service.py had this bug latent.
  • The HTTP 500 had nothing to do with the original IntegrityError; debugging required reading the gunicorn error log line by line.

Two helpers, two layers

Helper Location Returns Use in
commit_or_rollback() utils/db_transaction.py bool (True = success) service layer, background jobs, anywhere that doesn't return a Flask response
safe_commit() utils/safe_commit.py Flask response | None api/v2/* handlers — returns error_response('...', 500) directly on failure

How to use — service layer

from utils.db_transaction import commit_or_rollback

def link_mtls_certificate(user, certificate):
    user.mtls_certificate_id = certificate.id
    if not commit_or_rollback():
        # Already rolled back. Logged with exc_info=True. Caller decides what to do.
        return False
    return True

How to use — API handler

from utils.safe_commit import safe_commit

@bp.route('/things', methods=['POST'])
@require_auth(['write:things'])
@require_json_body
def create_thing():
    thing = Thing(**request.json)
    db.session.add(thing)
    if not safe_commit():
        return error_response('Failed to create', 500)
    return created_response(data=thing.to_dict())

Test coverage

backend/tests/test_db_transaction.py — 5 tests covering success, IntegrityError rollback, double-call idempotency.

Symmetric encryption at rest — security.encryption

UCM stores three categories of secret in the database:

Category Helper pair Wire format
Private key blobs (CA.prv, Certificate.prv) — base64 input encrypt_private_key() / decrypt_private_key() b64(MARKER + Fernet token of base64-input)
Text/PEM/JSON secrets (ACME proxy key, OIDC client secret, webhook secret) encrypt_text() / decrypt_text() (v2.144+) b64(MARKER + Fernet token of utf-8 text)
⚠️ Legacy encrypt() / decrypt() and encrypt_string() / decrypt_string() Don't use directly in new code —

The #105 regression

encrypt() expects base64-encoded bytes as input (it's the worker for encrypt_private_key). When a site passed a raw PEM string to encrypt(), it was silently re-base64-encoded as if it were already base64, producing garbage on decrypt. Always match the helper to the input contract:

# PEM, JSON, plain text, OIDC tokens → encrypt_text / decrypt_text
encrypted_blob = encrypt_text(pem_string)
pem_back = decrypt_text(encrypted_blob)

# Already-base64 private key blobs (CA.prv, Certificate.prv) → encrypt_private_key / decrypt_private_key
ca.prv = encrypt_private_key(base64.b64encode(pem_bytes).decode())

Singleton + key rotation

KeyEncryption is a singleton (__new__-based). After mutating KEY_ENCRYPTION_KEY in the environment (tests, key rotation), call KeyEncryption().reload() — otherwise the cached Fernet instance keeps using the old key.

Test coverage

  • tests/test_pem_encryption_helpers.py — encrypt_text/decrypt_text round-trip
  • tests/test_acme_proxy_key_encrypted.py — #105 regression
  • tests/test_key_encryption_pem_passthrough.py — decrypt_private_key() tolerates plaintext PEM

Trusted-proxy gating — utils.trusted_proxy 

from utils.trusted_proxy import (
    is_request_from_trusted_proxy,   # bool — True only if request originates from a CIDR in security.trusted_proxies
    client_ip,                       # respects X-Forwarded-For ONLY behind trusted proxy, else remote_addr
    reject_untrusted_proxy_headers,  # 401 helper for mTLS/EST/SCEP routes
)

Use client_ip() for every audit log. Reading request.remote_addr directly when ProxyFix is in the chain logs the proxy IP, not the user's. Reading X-Forwarded-For directly without checking trust lets unauthenticated callers spoof their IP in the audit log.

Use reject_untrusted_proxy_headers() for any route that consumes proxy-injected client cert headers (X-SSL-Client-Cert, X-SSL-Client-Verify, etc.) — EST, SCEP, mTLS login. Direct hits without TLS termination by a trusted proxy must be rejected.

SSRF outbound checks — utils.ssrf_protection

UCM is a LAN-deployed PKI. RFC1918, loopback, .lan / .local / .corp are the primary use case, not an attack vector.

Helper Blocks Use for
validate_url_not_cloud_metadata cloud metadata IPs (AWS/GCP/Azure/Alibaba) + loopback only ✅ Default for any user-supplied URL/host
validate_url_not_private private + loopback + reserved + link-local ❌ Almost never. Only for hosts that must be public Internet.

The v2.124 release accidentally swapped to validate_url_not_private for webhooks, ACME local validation, and OPNsense import — breaking every internal use case. Fixed in v2.126. Don't repeat that mistake. Checklist when adding any outbound feature:

  • Use validate_url_not_cloud_metadata, not validate_url_not_private.
  • Test with an RFC1918 target.
  • Test with a .local / .lan hostname.
  • Confirm cloud metadata IP (http://169.254.169.254) is still blocked.

API response shape — utils.response 

from utils.response import success_response, error_response, created_response, no_content_response

return success_response(data=result, message="Optional message")
return error_response("Generic message — never expose internals", 400)
return created_response(data=new_item)
return no_content_response()        # 204 — takes NO parameters

Never call jsonify(...) directly in api/v2/*. Never echo exception text to the client (error_response(str(e), 500) is a leak — log with exc_info=True, return a generic message).

Default-deny opt-in pattern

Some operations are disabled by default and only enabled by an explicit operator-set environment variable. They return 403 with a hint message naming the env var.

Feature Env var Endpoint
Runtime HSM pip install UCM_ALLOW_RUNTIME_PIP=1 POST /api/v2/hsm/install-dependencies

Template for new "danger" features (running shell, mutating system packages, executing user code):

import os
ALLOW = os.environ.get('UCM_ALLOW_FEATURE_X', '').lower() in ('1', 'true', 'yes')

@bp.route(...)
@require_auth(['admin:system'])
def dangerous_op():
    if not ALLOW:
        return error_response(
            "Feature X is disabled. Set UCM_ALLOW_FEATURE_X=1 to opt in, "
            "or perform the operation via your system package manager.",
            403
        )
    ...

Tests must monkeypatch the env var and the module-level constant if it's read at import time:

def test_dangerous_allowed(client, monkeypatch):
    monkeypatch.setenv('UCM_ALLOW_FEATURE_X', '1')
    monkeypatch.setattr('api.v2.hsm.ALLOW', True)

Pre-commit checklist

Before opening a PR that touches backend/:

  • No new base64.b64decode(decrypt_private_key(...)) — use utils.key_codec.load_pem_bytes().
  • No new bare db.session.commit() — use commit_or_rollback() or safe_commit().
  • No new except Exception: pass — at minimum logger.warning("...", exc_info=True).
  • No error_response(str(e), 500) — generic message, log with exc_info=True.
  • Audit log call uses client_ip(), not request.remote_addr.
  • If outbound to user-supplied URL: validate_url_not_cloud_metadata, never validate_url_not_private.
  • If new auth method: returns {user, role, permissions, csrf_token} (the full contract — see Auth Response Contract in Architecture).
  • If new public protocol endpoint: exempted from CSRF, FQDN redirect, HTTPS redirect, safe-mode check, SPA catch-all.
  • Backend suite green (cd backend && pytest tests/ -x -q).
  • If frontend touched: frontend suite green (cd frontend && npm test).

See also

Clone this wiki locally