docs(v1.9.0): correct upgrade notes on built-in role re-seed + backfill 1.8.8-1.8.10 release notes

Found during a v1.8.10 to v1.9.0 upgrade drill on a populated database
(schema v9 to v10) before releasing. Documentation only, no code change.

1. The v1.9.0 upgrade notes said "custom roles need no changes", which reads
   as "role data is untouched". It is not: because the SCHEMA_VERSION bump
   re-runs the whole idempotent sequence, update_system_roles_to_enterprise_rbac()
   issues an unconditional UPDATE roles SET ... permissions = <defaults> for
   the four BUILT-IN roles. In the drill an `operator` role that had been
   narrowed by removing apply.execute and config.bulk_import came back with
   both restored (57 to 59 permissions). Operator-created roles are NOT
   affected; the re-seed matches the four built-in names only.

   This is pre-existing behaviour of every SCHEMA_VERSION bump and is
   documented as intentional in migrations.py, so it is not introduced by the
   CSR feature. The v1.7.0 upgrade notes carried this caveat and it was not
   carried forward. Restored, with an export and re-apply procedure.

   Also clarified why the admin password is safe: the default-user seeding is
   guarded by an existence check ("safer than ON CONFLICT"), not an upsert,
   so an operator-changed password survives.

2. README release notes jumped from v1.8.7 straight to v1.9.0 because
   v1.8.8, v1.8.9 and v1.8.10 were never backfilled. Added all three.

README.md

@@ -105,6 +105,7 @@ This architecture provides better security (no inbound connections to HAProxy se
 ✅ **Version Control & Rollback** - Every change versioned with one-click restore capability  
 ✅ **Real-Time Monitoring** - Live stats, health checks, and performance dashboards  
 ✅ **SSL Certificate Management** - Centralized SSL with expiration tracking  
+✅ **CSR Creation** *(v1.9.0)* - Generate a private key + CSR in-app (RSA 2048/4096, ECDSA P-256/P-384, full subject + SANs), have it signed by any external CA, then import the signed certificate — the key never leaves the server  
 ✅ **ACME Auto SSL (Let's Encrypt)** - Automated certificate issuance, renewal, and deployment via ACME protocol  
 ✅ **ACME DNS-01 Challenge** *(v1.8.0)* - TXT-record validation for internal/isolated clusters (no public port 80) and wildcard certificates; pluggable DNS providers (Manual + Cloudflare), opt-in, HTTP-01 unchanged  
 ✅ **ACME Certificate Diagnostic Panel** - Automated preflight that checks agent readiness, DNS resolution, port 80 reachability, and ACME challenge ACL before issuing certificates  
@@ -778,6 +779,15 @@ User Updates SSL in UI → All Agents Poll Backend (30s)
   → Validate Config → Reload HAProxy (zero downtime)
 ```
 
+#### CSR Workflow (external / corporate CAs) — v1.9.0
+
+For certificates signed by an external or corporate CA, the **CSR tab** on the SSL Certificates page covers the whole flow without the private key ever leaving the server:
+
+1. **Create CSR**: pick a name (becomes the certificate name / on-agent file path), Common Name, optional SANs and subject fields (O/OU/L/ST/C/email), and a key algorithm (RSA 2048/4096 or ECDSA P-256/P-384). The backend generates the key + CSR; only the CSR PEM is shown (copy or download as `.csr`).
+2. **Get it signed**: submit the CSR to your Certificate Authority.
+3. **Import**: paste the signed certificate (+ optional chain), choose Global or cluster-specific scope and usage type. The backend verifies the certificate matches the stored key, rejects expired certs, warns on SAN drift, and creates a normal SSL certificate entry (source: `CSR`).
+4. **Deploy**: the imported certificate goes through the standard **PENDING → Apply Management → agent pull** pipeline like any other certificate.
+
 #### Key Features
 - **Certificate Upload**: PEM format certificate and private key upload
 - **ACME Automation**: Automatic certificate issuance and renewal via Let's Encrypt / ACME protocol (see [ACME Auto SSL](#acme-auto-ssl---automated-certificate-management))
@@ -1856,6 +1866,42 @@ GET /api/backends?cluster_id=1
 GET /api/frontends?cluster_id=1
 ```
 
+### SSL CSR API (v1.9.0)
+```bash
+# Create a CSR (generates the private key server-side; response contains the
+# CSR PEM — the private key is never returned by any endpoint)
+POST /api/ssl/csrs
+Authorization: Bearer <token>
+{
+  "name": "www-example-com",
+  "common_name": "www.example.com",
+  "sans": ["api.example.com"],
+  "key_algorithm": "rsa-2048",   # rsa-2048 | rsa-4096 | ecdsa-p256 | ecdsa-p384
+  "organization": "Example Corp",
+  "country": "TR"
+}
+
+# List CSRs (metadata only, no PEM)
+GET /api/ssl/csrs
+
+# CSR detail (includes the CSR PEM)
+GET /api/ssl/csrs/{csr_id}
+
+# Import the CA-signed certificate for a pending CSR
+POST /api/ssl/csrs/{csr_id}/import
+{
+  "certificate_content": "-----BEGIN CERTIFICATE-----...",
+  "chain_content": "-----BEGIN CERTIFICATE-----...",   # optional
+  "usage_type": "frontend",                            # frontend | server
+  "is_global": false,
+  "cluster_ids": [1, 2]
+}
+
+# Delete a CSR (pending: permanently destroys the private key;
+# completed: removes history only — the imported certificate is unaffected)
+DELETE /api/ssl/csrs/{csr_id}
+```
+
 ### ACME / Let's Encrypt API
 ```bash
 # List ACME accounts
@@ -2428,6 +2474,10 @@ Developed with ❤️ for the HAProxy community
 
 ## Release Notes
 
+- **v1.9.0** (2026-08-04) — **CSR creation** (in-app key + CSR generation and signed-certificate import): a new **CSR tab** on the SSL Certificates page generates a private key and Certificate Signing Request server-side (RSA 2048/4096 or ECDSA P-256/P-384; full subject — O/OU/L/ST/C/email — plus DNS SANs with wildcard support), for certificates signed by an **external or corporate CA**. The operator downloads/copies the CSR PEM, has it signed, then imports the signed certificate (+ optional chain): the backend verifies the certificate against the stored key (hard gate), rejects expired certs, warns on SAN drift, and creates a normal SSL certificate entry (source `CSR`) that flows through the standard **PENDING → Apply Management → agent pull** pipeline. The private key **never leaves the server** — no CSR endpoint returns it, and after import the CSR row's key copy is destroyed (the key then lives only on the certificate, like every other key). Additive schema change: one new table `ssl_csrs` (SCHEMA_VERSION 9 → 10, auto-migrated, no existing table altered); key generation runs off the event loop and is rate-limited per user; existing `ssl.*` permissions govern all new endpoints. No agent or rendered-config changes.
+- **v1.8.10** (2026-07-20) — **Security hardening** (GHSA-7rhv-c5pc-69r8, GHSA-3p5c-m5m4-mjpx, GHSA-3vh4): three advisory classes remediated, backend-only, no agent changes. (1) **RCE**: the agent script-template read/write endpoints now require the `agents.version` permission on top of authentication — a poisoned template is executed as root on every HAProxy node, so authentication alone was insufficient. (2) **Missing authentication**: operator/UI endpoints that were served without a JWT (dashboard stats, pool/cluster listings, agent inventory, WAF rules, config validate/optimize, SSL config-versions, health deep/agents/clusters) are now gated by a `require_authenticated_user` dependency, and agent data-plane endpoints that treated the `X-API-Key` header as *optional* (heartbeat, config, ssl-certificates, upgrade-status, pending-requests) now hard-reject a missing key. In every case the auth check was moved **ahead of** the handler's `try:` block so a 401 can no longer be rewritten into a 500 by the generic exception handler. (3) **SSRF**: a new `utils/ssrf_guard.py` (https-only, IPv4-pinned connector, all resolved addresses must be public, no redirects) protects the ACME directory fetch, the signed-request target and the ACME connection test, which accept operator- or DB-supplied URLs; the connection test also stopped reflecting arbitrary upstream JSON. Frontend dependency advisories patched in the same release. No schema, API-shape or rendered-config changes.
+- **v1.8.9** (2026-07-13) — **ACL `-f` pattern-file support** (Issue #38 follow-up): ACL definitions that reference a host-side pattern file (`acl … -f /etc/haproxy/lists/blocked.lst`) are accepted on import and edit instead of being rejected. The referenced file lives on the HAProxy node and cannot be validated from the manager, so the manager emits an **advisory warning** rather than a hard rejection and lets the agent's `haproxy -c` check be the fail-safe gate (a broken reference fails validation on the node and the previous config is restored). Consistent with the SPOE handling introduced in v1.8.8.
+- **v1.8.8** (2026-07-10) — **SPOE filter and frontend `log-format` preserved on import/edit** (Issue #38): importing an existing `haproxy.cfg` or editing a frontend silently dropped `filter spoe …` directives and custom `log-format` lines, so the next Apply pushed a config that had lost them. Both are now round-tripped through import and edit. As with `-f` pattern files, the SPOE engine config is a host-side file the manager cannot read, so it is preserved verbatim and reported as an advisory rather than validated centrally.
 - **v1.8.7** (2026-07-09) — **Version reporting single-source fix**: the version shown in the UI (backend-sourced via `/api/version`) could lag behind the real release. The canonical version lived in the repo-root `version.json`, but the backend image is built from the `./backend` context, so that file did not reach the container in every pipeline; the backend then fell back to a hardcoded constant in `main.py` that had to be bumped by hand and had drifted (it reported 1.8.4 after 1.8.5/1.8.6 shipped). The version now lives in a single file, `backend/version.json`, baked into every image automatically, and `main.py` no longer carries a real version literal (its fallback is a neutral "unknown"). A new test enforces that the version stays single-source and cannot drift. No functional or API change.
 - **v1.8.6** (2026-07-06) — **Performance: opt-in API workers + heartbeat micro-optimization** (Issue #35 follow-up): the backend container can now run multiple uvicorn worker processes via the new `UVICORN_WORKERS` environment variable (default **1** — behavior unchanged unless you opt in), letting the API use all cores on multi-core hosts; background tasks were already multi-replica safe, as exercised by the Kubernetes HPA deployment. The agent heartbeat handler now reads the agent's `status`/`version`/`upgrade_status` in one query instead of three (one round-trip per heartbeat, per agent, every 30s). Added a *Performance Tuning* section to the README (worker/replica scaling and how to use the `X-Response-Time` header and `Slow request detected` logs to pinpoint slow endpoints). Zero-risk release: no schema, API, or agent changes; defaults preserve existing behavior exactly.
 - **v1.8.5** (2026-07-03) — **ACME completion-task SQL fix** (Issue #35 follow-up): the background order-completion task (`complete_pending_acme_orders`, runs every 60s) died on **every cycle** with `syntax error at or near ")"` — an extra closing parenthesis introduced in v1.8.0's bounded DNS-01 retry claim query. Because that query is the task's first database call, **no background ACME work ran at all from v1.8.0 through v1.8.4**: orders were never claimed for finalize/download, the DNS-01 TXT record was never published (so DNS-01 with an automated provider such as Cloudflare could never validate), Site Wizard staged orders never left `wizard_staged`, and DNS-01 retry/TXT-cleanup never executed. The stray parenthesis is removed and a regression test now scans all ACME modules' SQL for unbalanced parentheses (the unit suite mocks the database, which is why a raw-SQL syntax error could slip through). One-line backend query fix; no schema, API, or agent changes — fully backward compatible.

UPGRADE_GUIDE.md

@@ -1,3 +1,46 @@
+# Upgrade Notes — v1.9.0 (CSR creation)
+
+**Backward compatible & additive.** Upgrading to v1.9.0 changes nothing for existing
+clusters/agents until you create a CSR:
+
+- **Schema:** `SCHEMA_VERSION` bumps to `10`, so on first start the (idempotent)
+  migration sequence re-runs once and adds **one new table** (`ssl_csrs`) plus its
+  indexes. **No existing table is altered**, existing rows are untouched, and the
+  **admin password is not reset** (the default-user seeding is guarded by an
+  existence check, not an upsert). No new permission strings are introduced — all
+  CSR endpoints are governed by the existing `ssl.create` / `ssl.read` /
+  `ssl.delete` permissions.
+- **⚠️ Built-in roles are re-seeded to their defaults (pre-existing behaviour of
+  every `SCHEMA_VERSION` bump — verified in a v1.8.10 → v1.9.0 upgrade drill).**
+  Because the version gate re-runs the whole sequence, `update_system_roles_to_enterprise_rbac()`
+  issues an unconditional `UPDATE roles SET … permissions = <defaults> WHERE name = …`
+  for the four **built-in** roles (`super_admin`, `operator`, `security_admin`,
+  `viewer`). **Any customization you made to a built-in role is reverted.** In the
+  drill, an `operator` role that had been narrowed by removing `apply.execute` and
+  `config.bulk_import` came back with both restored (57 → 59 permissions).
+  - **Roles you created yourself are NOT affected** — the re-seed matches on the four
+    built-in names only.
+  - This is not new in v1.9.0: it happens on every release that bumps
+    `SCHEMA_VERSION` (v1.7.0, v1.8.0, v1.8.8 …). It is documented as intentional at
+    `backend/database/migrations.py` (the "BUMP THIS … OR seeded/role data" note) —
+    the migration is treated as the authority on built-in-role contents.
+  - **If you have hardened a built-in role, do this:** export it before upgrading
+    (`GET /api/roles`), then re-apply your changes after the first start
+    (`PUT /api/roles/{id}`) — or, preferably, move your customization into a
+    purpose-made custom role, which survives every upgrade.
+- **Key storage:** CSR private keys are stored in the database like every other key
+  in the system (`ssl_certificates.private_key_content` and the ACME order keys).
+  The key is never returned by any CSR API endpoint, and after a successful import
+  the CSR row's key copy is set to NULL (the key then lives only on the certificate
+  row).
+- **Agents:** zero agent changes. Agents never read the new table; a CSR becomes
+  visible to agents only after its signed certificate is imported **and** applied via
+  Apply Management (the standard PENDING pipeline).
+- **Rollback:** simply don't use the CSR tab. The `ssl_csrs` table is inert when
+  empty; downgrading the application leaves it as an ignored extra table.
+
+---
+
 # Upgrade Notes — v1.7.0 (HA / VIP Keepalived management, Issue #27)
 
 **Backward compatible & opt-in.** Upgrading to v1.7.0 changes nothing for existing

backend/database/migrations.py

@@ -1753,7 +1753,12 @@ async def ensure_agent_activity_logs_table():
 # bump, already-deployed databases (version >= 8) skip the whole migration run and never gain
 # the columns, so the frontends SELECT/INSERT would fail. Additive + idempotent + nullable;
 # existing rows stay NULL and render byte-identical.
-SCHEMA_VERSION = 9
+# v1.9.0 (CSR creation): bumped 9 -> 10 for the brand-new `ssl_csrs` table
+# (ensure_ssl_csrs_table step). Holds a locally generated private key + CSR PEM
+# until the operator imports the CA-signed certificate; the import creates a
+# normal ssl_certificates row and NULLs the key copy here. Additive + idempotent;
+# no existing table is altered, agents never read this table.
+SCHEMA_VERSION = 10
 
 
 async def run_all_migrations():
@@ -1890,12 +1895,84 @@ async def _run_all_migrations_inner():
     await ensure_mfa_columns()
 
     # Issue #27 — HA/VIP Keepalived management (v1.7.0): two brand-new tables.
-    # MUST stay last: FK-references haproxy_cluster_pools/agents/users, all created above.
+    # MUST run after its FK targets (haproxy_cluster_pools/agents/users), all created above.
     await ensure_vip_tables()
 
+    # v1.9.0 — CSR creation: brand-new ssl_csrs table. FK-references
+    # ssl_certificates/users, both created above.
+    await ensure_ssl_csrs_table()
+
     logger.info("Database migrations completed successfully.")
 
 
+async def ensure_ssl_csrs_table():
+    """v1.9.0 — CSR (Certificate Signing Request) creation. Additive only:
+    one brand-new table (ssl_csrs) + indexes. No ALTER of any existing table,
+    so the entire current fleet is byte-identical. Fully idempotent
+    (CREATE TABLE/INDEX IF NOT EXISTS). FK targets (ssl_certificates, users)
+    are created earlier in the sequence.
+
+    A CSR row holds a locally generated private key + CSR PEM until the
+    operator imports the CA-signed certificate. The import creates a normal
+    ssl_certificates row (source='csr', last_config_status='PENDING') and
+    NULLs the private_key_pem copy here — the key then lives only on the
+    certificate row, like every other key in the system. Agents never read
+    this table: the agent SSL delivery endpoint selects from
+    ssl_certificates only, so a pending CSR can never leak to an agent.
+    """
+    conn = None
+    try:
+        conn = await get_database_connection()
+
+        await conn.execute("""
+            CREATE TABLE IF NOT EXISTS ssl_csrs (
+                id SERIAL PRIMARY KEY,
+                name VARCHAR(100) NOT NULL,
+                common_name VARCHAR(253) NOT NULL,
+                subject JSONB NOT NULL DEFAULT '{}'::jsonb,
+                sans JSONB NOT NULL DEFAULT '[]'::jsonb,
+                key_algorithm VARCHAR(20) NOT NULL DEFAULT 'rsa-2048',
+                csr_pem TEXT NOT NULL,
+                private_key_pem TEXT,
+                status VARCHAR(20) NOT NULL DEFAULT 'pending',
+                ssl_certificate_id INTEGER REFERENCES ssl_certificates(id) ON DELETE SET NULL,
+                completed_at TIMESTAMP,
+                created_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                CONSTRAINT ssl_csrs_status_check CHECK (status IN ('pending', 'completed'))
+            );
+        """)
+
+        # Only PENDING CSRs reserve their name: the name becomes the
+        # ssl_certificates.name (and thus /etc/ssl/haproxy/{name}.pem on every
+        # agent) at import time, so two open CSRs must not target the same
+        # cert name. Completed CSRs are history and may share a name across
+        # reissues — mirrors the uq_vip_name_active partial-index rationale.
+        await conn.execute(
+            "CREATE UNIQUE INDEX IF NOT EXISTS uq_ssl_csrs_name_pending ON ssl_csrs(name) WHERE status = 'pending';"
+        )
+        await conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_ssl_csrs_status ON ssl_csrs(status);"
+        )
+        await conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_ssl_csrs_cert ON ssl_csrs(ssl_certificate_id);"
+        )
+
+        logger.info("ssl_csrs table ensured (v1.9.0 CSR creation)")
+    except Exception as e:
+        logger.error(f"Error ensuring ssl_csrs table: {e}")
+        # Re-raise (ensure_ssl_cluster_junction_table precedent): this step is
+        # part of the SCHEMA_VERSION=10 bump, and run_all_migrations() records
+        # the marker only after the inner sequence completes cleanly. Swallowing
+        # a failure here would stamp version 10 with no ssl_csrs table, and the
+        # version gate would then skip every future retry — permanently.
+        raise
+    finally:
+        if conn:
+            await close_database_connection(conn)
+
+
 async def ensure_mfa_columns():
     """Issue #18 — TOTP MFA (v1.6.0): additive columns on users + 3 new tables.
 

backend/main.py

@@ -47,6 +47,7 @@ from routers.acme_diagnostics import router as acme_diagnostics_router
 from routers.site_wizard import router as site_wizard_router
 from routers.mfa import router as mfa_router
 from routers.vip import router as vip_router  # Issue #27 — HA/VIP (Keepalived) management
+from routers.csr import router as csr_router  # v1.9.0 — CSR creation (in-app key+CSR generation, signed-cert import)
 
 # Production logging configuration
 from utils.logging_config import setup_production_logging
@@ -892,6 +893,7 @@ app.include_router(dashboard_stats_router)  # HAProxy stats dashboard
 app.include_router(agent_router)
 app.include_router(waf_router)
 app.include_router(ssl_router)
+app.include_router(csr_router)  # v1.9.0: CSR creation (in-app key+CSR generation, signed-cert import)
 app.include_router(security_router)
 app.include_router(configuration_router)
 app.include_router(settings_router)

backend/models/csr.py

@@ -0,0 +1,251 @@
+"""
+Pydantic models for the CSR (Certificate Signing Request) feature (v1.9.0).
+
+A CSR row is the precursor of an ssl_certificates row: the backend generates
+the private key + CSR locally, the operator has the CSR signed by an external
+CA and then imports the signed certificate. The CSR `name` therefore obeys the
+exact same path-traversal contract as the SSL certificate name (Bulgu #21) —
+at import time it becomes /etc/ssl/haproxy/{name}.pem on every agent and is
+shell-processed by the agent script as root.
+
+The import model deliberately has NO private key field: the key never leaves
+the server. It is stored on the ssl_csrs row at generation time and paired
+with the signed certificate server-side.
+"""
+
+import re
+from typing import List, Optional
+
+from pydantic import BaseModel, field_validator, model_validator
+
+KEY_ALGORITHMS = ('rsa-2048', 'rsa-4096', 'ecdsa-p256', 'ecdsa-p384')
+
+# RFC 1035 LDH hostname, lowercase, optional single leftmost wildcard label.
+# Single-label names are allowed (internal CAs routinely sign bare hostnames).
+_DNS_NAME_PATTERN = re.compile(
+    r'^(\*\.)?[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?'
+    r'(\.[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?)*$'
+)
+
+# Reject control characters in free-text subject fields: they would be
+# persisted, echoed into the UI / issuer column, and printed into agent logs
+# via `openssl -subject` output.
+_CONTROL_CHARS_PATTERN = re.compile(r'[\x00-\x1f\x7f]')
+
+_MAX_SANS = 100
+_MAX_CERT_PEM_BYTES = 64 * 1024       # a leaf certificate is ~2 KB; 64 KB is generous
+_MAX_CHAIN_PEM_BYTES = 256 * 1024     # agents re-download all cert content every poll
+
+
+def _validate_dns_name(value: str, field_label: str) -> str:
+    v = (value or '').strip().lower()
+    if not v:
+        raise ValueError(f'{field_label} must not be empty')
+    if len(v) > 253:
+        raise ValueError(f'{field_label} must be 253 characters or fewer')
+    if not _DNS_NAME_PATTERN.match(v):
+        raise ValueError(
+            f'{field_label} {value!r} is not a valid DNS name — lowercase '
+            'letters, digits, hyphens and dots only; a wildcard is allowed '
+            'only as the leftmost label (e.g. *.example.com).'
+        )
+    return v
+
+
+def _validate_subject_text(value: Optional[str], field_label: str, max_len: int = 64) -> Optional[str]:
+    if value is None:
+        return None
+    v = value.strip()
+    if not v:
+        return None
+    if len(v) > max_len:
+        raise ValueError(f'{field_label} must be {max_len} characters or fewer')
+    if _CONTROL_CHARS_PATTERN.search(v):
+        raise ValueError(f'{field_label} must not contain control characters')
+    return v
+
+
+def _validate_csr_name(v: str) -> str:
+    """Mirror of SSLCertificateCreate.validate_name_no_path_traversal (Bulgu #21)
+    with one deliberate tightening: max length 100, matching the
+    ssl_certificates.name VARCHAR(100) column (the historical 200-char limit
+    overflows the column and 500s — not replicated here)."""
+    if v is None:
+        raise ValueError('CSR name is required')
+    stripped = v.strip()
+    if not stripped:
+        raise ValueError('CSR name must not be empty')
+    if stripped != v:
+        raise ValueError('CSR name must not contain leading/trailing whitespace')
+    if len(stripped) > 100:
+        raise ValueError('CSR name must be 100 characters or fewer')
+    if not re.match(r'^[A-Za-z0-9_.-]+$', stripped):
+        raise ValueError(
+            f'CSR name={v!r} contains forbidden characters — only letters, '
+            'digits, underscore, hyphen, and dot are allowed (the name becomes '
+            'a filename component under /etc/ssl/haproxy/ at import).'
+        )
+    if '..' in stripped:
+        raise ValueError(f'CSR name={v!r} must not contain ".." (path traversal)')
+    if stripped.startswith('.'):
+        raise ValueError(f'CSR name={v!r} must not start with "." (hidden filename)')
+    if stripped.startswith('-'):
+        raise ValueError(f'CSR name={v!r} must not start with "-" (CLI flag confusion)')
+    return stripped
+
+
+class SSLCSRCreate(BaseModel):
+    name: str                                   # becomes the certificate name at import
+    common_name: str
+    organization: Optional[str] = None          # O
+    organizational_unit: Optional[str] = None   # OU
+    locality: Optional[str] = None              # L
+    state: Optional[str] = None                 # ST
+    country: Optional[str] = None               # C — exactly 2 letters
+    email: Optional[str] = None                 # emailAddress
+    sans: List[str] = []                        # DNS names; CN is auto-added server-side
+    key_algorithm: str = 'rsa-2048'
+
+    @field_validator('name')
+    @classmethod
+    def validate_name(cls, v):
+        return _validate_csr_name(v)
+
+    @field_validator('common_name')
+    @classmethod
+    def validate_common_name(cls, v):
+        v = _validate_dns_name(v, 'Common Name')
+        # RFC 5280 ub-common-name — many CAs reject CNs longer than 64 chars.
+        if len(v) > 64:
+            raise ValueError(
+                'Common Name must be 64 characters or fewer (RFC 5280 upper '
+                'bound) — put longer names in the SAN list instead.'
+            )
+        return v
+
+    @field_validator('sans')
+    @classmethod
+    def validate_sans(cls, v):
+        if not v:
+            return []
+        if len(v) > _MAX_SANS:
+            raise ValueError(f'At most {_MAX_SANS} SAN entries are allowed')
+        seen = set()
+        result = []
+        for entry in v:
+            normalised = _validate_dns_name(entry, 'SAN entry')
+            if normalised not in seen:
+                seen.add(normalised)
+                result.append(normalised)
+        return result
+
+    @field_validator('organization')
+    @classmethod
+    def validate_organization(cls, v):
+        return _validate_subject_text(v, 'Organization (O)')
+
+    @field_validator('organizational_unit')
+    @classmethod
+    def validate_organizational_unit(cls, v):
+        return _validate_subject_text(v, 'Organizational Unit (OU)')
+
+    @field_validator('locality')
+    @classmethod
+    def validate_locality(cls, v):
+        return _validate_subject_text(v, 'Locality (L)')
+
+    @field_validator('state')
+    @classmethod
+    def validate_state(cls, v):
+        return _validate_subject_text(v, 'State/Province (ST)')
+
+    @field_validator('country')
+    @classmethod
+    def validate_country(cls, v):
+        # cryptography raises a bare ValueError for a non-2-char COUNTRY_NAME;
+        # pre-validate so the operator gets a friendly 422 instead of a 500.
+        if v is None:
+            return None
+        v = v.strip()
+        if not v:
+            return None
+        if not re.match(r'^[A-Za-z]{2}$', v):
+            raise ValueError('Country (C) must be exactly 2 letters (ISO 3166-1 alpha-2, e.g. TR, US)')
+        return v.upper()
+
+    @field_validator('email')
+    @classmethod
+    def validate_email(cls, v):
+        v = _validate_subject_text(v, 'Email', max_len=254)
+        if v is not None and ('@' not in v or v.startswith('@') or v.endswith('@')):
+            raise ValueError('Email must be a valid address (missing or misplaced "@")')
+        return v
+
+    @field_validator('key_algorithm')
+    @classmethod
+    def validate_key_algorithm(cls, v):
+        if v not in KEY_ALGORITHMS:
+            raise ValueError(
+                f'key_algorithm must be one of: {", ".join(KEY_ALGORITHMS)}'
+            )
+        return v
+
+
+class SSLCSRImport(BaseModel):
+    """Import the CA-signed certificate for a pending CSR. The private key is
+    NOT part of the request — it is already stored on the CSR row."""
+    certificate_content: str                    # PEM
+    chain_content: Optional[str] = None         # PEM, optional
+    usage_type: str = 'frontend'                # "frontend" or "server"
+    is_global: bool = False
+    cluster_ids: Optional[List[int]] = None
+    # Escape hatch for name collisions that appeared AFTER the CSR was
+    # created: overrides the CSR's reserved name for the certificate row.
+    name: Optional[str] = None
+
+    @field_validator('certificate_content')
+    @classmethod
+    def validate_certificate(cls, v):
+        if not v or not v.strip():
+            raise ValueError('Certificate content is required')
+        v = v.strip()
+        if len(v.encode('utf-8', errors='ignore')) > _MAX_CERT_PEM_BYTES:
+            raise ValueError('Certificate content exceeds the 64 KB limit')
+        if '-----BEGIN CERTIFICATE-----' not in v or '-----END CERTIFICATE-----' not in v:
+            raise ValueError('Certificate must be in PEM format')
+        return v
+
+    @field_validator('chain_content')
+    @classmethod
+    def validate_chain(cls, v):
+        if v and v.strip():
+            v = v.strip()
+            if len(v.encode('utf-8', errors='ignore')) > _MAX_CHAIN_PEM_BYTES:
+                raise ValueError('Certificate chain exceeds the 256 KB limit')
+            if '-----BEGIN CERTIFICATE-----' not in v or '-----END CERTIFICATE-----' not in v:
+                raise ValueError('Certificate chain must be in PEM format')
+            return v
+        return None
+
+    @field_validator('usage_type')
+    @classmethod
+    def validate_usage_type(cls, v):
+        if v not in ['frontend', 'server']:
+            raise ValueError('usage_type must be either "frontend" or "server"')
+        return v
+
+    @field_validator('name')
+    @classmethod
+    def validate_name(cls, v):
+        if v is None or not str(v).strip():
+            return None
+        return _validate_csr_name(v)
+
+    @model_validator(mode='after')
+    def validate_cluster_selection(self):
+        if not self.is_global and not self.cluster_ids:
+            raise ValueError(
+                'cluster_ids is required when is_global is false — pick at '
+                'least one cluster or import the certificate as global.'
+            )
+        return self

backend/routers/csr.py

@@ -0,0 +1,375 @@
+"""
+CSR (Certificate Signing Request) endpoints (v1.9.0).
+
+Generate a private key + CSR in-app, download the CSR PEM, have it signed by
+an external CA, then import the signed certificate — which creates a normal
+ssl_certificates row that flows through the existing pipeline
+(config version → Apply Management → agent pull).
+
+Security posture:
+- All endpoints enforce ssl.* permissions explicitly (including the read
+  endpoints — deliberately stricter than the legacy cert detail route).
+- The private key is NEVER returned by any endpoint here; after import it is
+  reachable only via the existing certificate detail route.
+- Key generation is offloaded to a thread (RSA-4096 takes seconds; the
+  backend runs a single-worker event loop by default) and rate-limited
+  per user via the user_activity_logs COUNT pattern (acme_diagnostics
+  precedent — slowapi is not registered on the app).
+"""
+
+import asyncio
+import logging
+from typing import Optional
+
+from fastapi import APIRouter, HTTPException, Request, Header
+
+from database.connection import get_database_connection, close_database_connection
+from auth_middleware import get_current_user_from_token, check_user_permission
+from models.csr import SSLCSRCreate, SSLCSRImport
+from services import csr_service, ssl_service
+from routers.ssl import _assert_safe_cert_name, validate_user_cluster_access
+from utils.activity_log import log_user_activity
+
+router = APIRouter(prefix="/api/ssl/csrs", tags=["SSL CSRs"])
+logger = logging.getLogger(__name__)
+
+_RATE_LIMIT_CREATE_PER_MIN = 10
+
+# Columns exposed to the API — private_key_pem is deliberately absent so a
+# future `SELECT *` refactor cannot silently start leaking it.
+_CSR_LIST_COLUMNS = """
+    c.id, c.name, c.common_name, c.subject, c.sans, c.key_algorithm,
+    c.status, c.ssl_certificate_id, c.completed_at, c.created_at, c.updated_at,
+    s.name AS certificate_name, u.username AS created_by_username
+"""
+
+_INT32_MAX = 2_147_483_647
+
+
+def _client_ip(request: Optional[Request]) -> Optional[str]:
+    try:
+        return str(request.client.host) if request and request.client else None
+    except Exception:
+        return None
+
+
+def _user_agent(request: Optional[Request]) -> Optional[str]:
+    try:
+        return request.headers.get("user-agent") if request else None
+    except Exception:
+        return None
+
+
+async def _require(authorization: Optional[str], action: str):
+    """Authenticate + enforce ssl.<action>; returns current_user or raises 401/403."""
+    current_user = await get_current_user_from_token(authorization)
+    ok = await check_user_permission(current_user["id"], "ssl", action, current_user=current_user)
+    if not ok:
+        raise HTTPException(status_code=403, detail=f"Insufficient permissions: ssl.{action} required")
+    return current_user
+
+
+def _assert_int32_id(csr_id: int) -> None:
+    """ssl_csrs.id is int4 — an out-of-range path param would surface as an
+    asyncpg DataError 500 (Bulgu #96 precedent); return a clean 404 instead."""
+    if csr_id < 1 or csr_id > _INT32_MAX:
+        raise HTTPException(status_code=404, detail="CSR not found")
+
+
+def _assert_valid_cluster_id(cluster_id: int) -> None:
+    """Same int4 guard for body-supplied cluster ids: haproxy_clusters.id is
+    SERIAL/int4, so an out-of-range value would raise asyncpg DataError inside
+    validate_user_cluster_access and surface as a 500 with the raw driver
+    error. Fail with the same clean 404 the cluster lookup itself produces."""
+    if not isinstance(cluster_id, int) or cluster_id < 1 or cluster_id > _INT32_MAX:
+        raise HTTPException(status_code=404, detail="Cluster not found")
+
+
+async def _enforce_create_rate_limit(conn, user_id: int) -> None:
+    """Per-user per-minute limit on key generation, counted against the
+    csr_create audit-log action (acme_diagnostics _enforce_rate_limit pattern,
+    backed by the (user_id, action, created_at DESC) composite index)."""
+    cnt = await conn.fetchval(
+        """
+        SELECT COUNT(*)
+        FROM user_activity_logs
+        WHERE user_id = $1
+          AND action = 'csr_create'
+          AND created_at >= NOW() - INTERVAL '60 seconds'
+        """,
+        user_id,
+    )
+    if cnt is not None and cnt >= _RATE_LIMIT_CREATE_PER_MIN:
+        raise HTTPException(
+            status_code=429,
+            detail=(
+                f"Rate limit exceeded: at most {_RATE_LIMIT_CREATE_PER_MIN} "
+                "CSRs may be created per minute"
+            ),
+        )
+
+
+@router.post("")
+async def create_csr(payload: SSLCSRCreate, request: Request, authorization: Optional[str] = Header(None)):
+    """Generate a private key + CSR. Returns the CSR PEM immediately (so the
+    UI can show copy/download in one round trip) — never the private key."""
+    current_user = await _require(authorization, "create")
+    conn = None
+    try:
+        # Belt and braces on top of the model validator — same duplication
+        # convention as the certificate create route.
+        _assert_safe_cert_name(payload.name)
+
+        conn = await get_database_connection()
+        await _enforce_create_rate_limit(conn, current_user["id"])
+
+        # Fail fast on a taken name BEFORE burning CPU on key generation;
+        # insert_csr_row re-checks and the partial unique index closes the race.
+        await csr_service.assert_csr_name_available(conn, payload.name)
+
+        bundle = await asyncio.to_thread(csr_service.generate_csr_bundle, payload)
+        csr_id = await csr_service.insert_csr_row(conn, payload, bundle, current_user["id"])
+
+        row = await conn.fetchrow(
+            f"""
+            SELECT {_CSR_LIST_COLUMNS}, c.csr_pem
+            FROM ssl_csrs c
+            LEFT JOIN ssl_certificates s ON c.ssl_certificate_id = s.id
+            LEFT JOIN users u ON c.created_by = u.id
+            WHERE c.id = $1
+            """,
+            csr_id,
+        )
+
+        await log_user_activity(
+            user_id=current_user["id"],
+            action='csr_create',
+            resource_type='ssl_csr',
+            resource_id=str(csr_id),
+            details={
+                'csr_name': payload.name,
+                'common_name': payload.common_name,
+                'sans': bundle['sans'],
+                'key_algorithm': payload.key_algorithm,
+            },
+            ip_address=_client_ip(request),
+            user_agent=_user_agent(request),
+        )
+
+        return {
+            "message": f"CSR '{payload.name}' created successfully",
+            "csr": csr_service.csr_row_to_dict(row, include_pem=True),
+        }
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error creating CSR: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        if conn:
+            await close_database_connection(conn)
+
+
+@router.get("")
+async def list_csrs(authorization: Optional[str] = Header(None)):
+    """List CSRs (no PEM payloads — fetch the detail route for the CSR PEM).
+    Cluster-agnostic: a CSR binds to clusters only at import time."""
+    await _require(authorization, "read")
+    conn = None
+    try:
+        conn = await get_database_connection()
+        rows = await conn.fetch(
+            f"""
+            SELECT {_CSR_LIST_COLUMNS}
+            FROM ssl_csrs c
+            LEFT JOIN ssl_certificates s ON c.ssl_certificate_id = s.id
+            LEFT JOIN users u ON c.created_by = u.id
+            ORDER BY c.created_at DESC
+            """
+        )
+        return [csr_service.csr_row_to_dict(r) for r in rows]
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error listing CSRs: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        if conn:
+            await close_database_connection(conn)
+
+
+@router.get("/{csr_id}")
+async def get_csr(csr_id: int, authorization: Optional[str] = Header(None)):
+    """CSR detail including the CSR PEM. The private key is never included."""
+    await _require(authorization, "read")
+    _assert_int32_id(csr_id)
+    conn = None
+    try:
+        conn = await get_database_connection()
+        row = await conn.fetchrow(
+            f"""
+            SELECT {_CSR_LIST_COLUMNS}, c.csr_pem
+            FROM ssl_csrs c
+            LEFT JOIN ssl_certificates s ON c.ssl_certificate_id = s.id
+            LEFT JOIN users u ON c.created_by = u.id
+            WHERE c.id = $1
+            """,
+            csr_id,
+        )
+        if not row:
+            raise HTTPException(status_code=404, detail="CSR not found")
+        return csr_service.csr_row_to_dict(row, include_pem=True)
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error fetching CSR {csr_id}: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        if conn:
+            await close_database_connection(conn)
+
+
+@router.post("/{csr_id}/import")
+async def import_csr_certificate(
+    csr_id: int,
+    payload: SSLCSRImport,
+    request: Request,
+    authorization: Optional[str] = Header(None),
+):
+    """Import the CA-signed certificate for a pending CSR. Creates an
+    ssl_certificates row (source='csr', PENDING) and stages one config
+    version per affected cluster — the operator applies manually."""
+    current_user = await _require(authorization, "create")
+    _assert_int32_id(csr_id)
+    conn = None
+    try:
+        if payload.name:
+            _assert_safe_cert_name(payload.name)
+
+        conn = await get_database_connection()
+
+        if not payload.is_global:
+            for cluster_id in payload.cluster_ids or []:
+                _assert_valid_cluster_id(cluster_id)
+                await validate_user_cluster_access(current_user["id"], cluster_id, conn)
+
+        result = await csr_service.import_signed_certificate(
+            conn, csr_id, payload, current_user["id"]
+        )
+        cert_id = result["certificate_id"]
+
+        if payload.is_global:
+            cluster_rows = await conn.fetch(
+                "SELECT id FROM haproxy_clusters WHERE is_active = TRUE"
+            )
+            affected_clusters = [r['id'] for r in cluster_rows]
+        else:
+            affected_clusters = payload.cluster_ids or []
+
+        # Post-commit staging — a config-generation failure never rolls back
+        # the certificate (same semantics as the manual create flow).
+        sync_results = await ssl_service.stage_ssl_config_versions(
+            conn, cert_id, affected_clusters, action='create',
+            created_by=current_user["id"],
+        )
+
+        await log_user_activity(
+            user_id=current_user["id"],
+            action='create',
+            resource_type='ssl_certificate',
+            resource_id=str(cert_id),
+            details={
+                'certificate_name': result['certificate_name'],
+                'domain': result.get('primary_domain', 'unknown'),
+                'via': 'csr',
+                'csr_id': csr_id,
+                'usage_type': payload.usage_type,
+                'is_global': payload.is_global,
+                'cluster_ids': payload.cluster_ids,
+                'warnings': result['warnings'],
+            },
+            ip_address=_client_ip(request),
+            user_agent=_user_agent(request),
+        )
+        await log_user_activity(
+            user_id=current_user["id"],
+            action='csr_import',
+            resource_type='ssl_csr',
+            resource_id=str(csr_id),
+            details={
+                'certificate_id': cert_id,
+                'certificate_name': result['certificate_name'],
+            },
+            ip_address=_client_ip(request),
+            user_agent=_user_agent(request),
+        )
+
+        return {
+            "message": (
+                f"Certificate '{result['certificate_name']}' imported "
+                "successfully. Go to Apply Management to deploy."
+            ),
+            "certificate_id": cert_id,
+            "warnings": result["warnings"],
+            "sync_results": sync_results,
+        }
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error importing signed certificate for CSR {csr_id}: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        if conn:
+            await close_database_connection(conn)
+
+
+@router.delete("/{csr_id}")
+async def delete_csr(csr_id: int, request: Request, authorization: Optional[str] = Header(None)):
+    """Hard delete. For a pending CSR this permanently destroys the private
+    key (any certificate later signed from that CSR becomes unusable); for a
+    completed CSR it only removes history — the imported certificate is not
+    affected (the FK points csr → cert)."""
+    current_user = await _require(authorization, "delete")
+    _assert_int32_id(csr_id)
+    conn = None
+    try:
+        conn = await get_database_connection()
+        async with conn.transaction():
+            # FOR UPDATE serialises against an in-flight import of the same CSR.
+            row = await conn.fetchrow(
+                "SELECT id, name, status FROM ssl_csrs WHERE id = $1 FOR UPDATE",
+                csr_id,
+            )
+            if not row:
+                raise HTTPException(status_code=404, detail="CSR not found")
+            await conn.execute("DELETE FROM ssl_csrs WHERE id = $1", csr_id)
+
+        await log_user_activity(
+            user_id=current_user["id"],
+            action='delete',
+            resource_type='ssl_csr',
+            resource_id=str(csr_id),
+            details={'csr_name': row['name'], 'status': row['status']},
+            ip_address=_client_ip(request),
+            user_agent=_user_agent(request),
+        )
+
+        if row['status'] == 'pending':
+            message = (
+                f"CSR '{row['name']}' deleted — its private key has been "
+                "permanently destroyed."
+            )
+        else:
+            message = (
+                f"CSR '{row['name']}' deleted (history only) — the imported "
+                "certificate is not affected."
+            )
+        return {"message": message}
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error deleting CSR {csr_id}: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+    finally:
+        if conn:
+            await close_database_connection(conn)

backend/services/csr_service.py

@@ -0,0 +1,432 @@
+"""
+csr_service: CSR (Certificate Signing Request) generation + signed-certificate
+import (v1.9.0).
+
+Flow:
+  1. `generate_csr_bundle` builds a private key + CSR locally (pure crypto,
+     no DB/IO — callers MUST run it via `asyncio.to_thread`: RSA-4096
+     generation takes seconds and would stall the single-worker event loop).
+  2. The bundle is persisted to `ssl_csrs` (`insert_csr_row`); the operator
+     downloads the CSR PEM and has it signed by an external CA.
+  3. `import_signed_certificate` pairs the CA response with the stored key,
+     creates a normal `ssl_certificates` row (source='csr',
+     last_config_status='PENDING' — agents never see it before Apply) and
+     NULLs the key copy on the CSR row.
+
+The CSR builder generalises the in-repo ACME reference
+(services/acme_service.py finalize_order): PEM output instead of DER, full
+subject instead of CN-only, ECDSA support, same PKCS8/NoEncryption key
+serialisation (the agent concatenates cert+key+chain into one PEM and HAProxy
+cannot read passphrase-protected keys).
+
+Private keys are stored PLAINTEXT, consistent with every other key in the
+system (ssl_certificates.private_key_content, letsencrypt_orders.cert_private_key).
+The key is NEVER returned by any CSR API response — `csr_row_to_dict` strips
+it unconditionally.
+"""
+
+import json
+import logging
+from typing import Any, Dict, List, Optional
+from types import SimpleNamespace
+
+import asyncpg
+from fastapi import HTTPException
+
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec, rsa
+from cryptography.x509.oid import NameOID
+
+from services import ssl_service
+
+logger = logging.getLogger(__name__)
+
+
+_KEY_FACTORIES = {
+    'rsa-2048': lambda: rsa.generate_private_key(public_exponent=65537, key_size=2048),
+    'rsa-4096': lambda: rsa.generate_private_key(public_exponent=65537, key_size=4096),
+    'ecdsa-p256': lambda: ec.generate_private_key(ec.SECP256R1()),
+    'ecdsa-p384': lambda: ec.generate_private_key(ec.SECP384R1()),
+}
+
+# (payload attribute, x509 OID, subject-JSON key)
+_SUBJECT_OID_MAP = [
+    ('organization', NameOID.ORGANIZATION_NAME, 'O'),
+    ('organizational_unit', NameOID.ORGANIZATIONAL_UNIT_NAME, 'OU'),
+    ('locality', NameOID.LOCALITY_NAME, 'L'),
+    ('state', NameOID.STATE_OR_PROVINCE_NAME, 'ST'),
+    ('country', NameOID.COUNTRY_NAME, 'C'),
+    ('email', NameOID.EMAIL_ADDRESS, 'emailAddress'),
+]
+
+
+def generate_csr_bundle(payload: Any) -> Dict[str, Any]:
+    """Generate a private key + CSR for a validated SSLCSRCreate payload.
+
+    Pure CPU-bound crypto — no DB, no network. Callers must offload via
+    `asyncio.to_thread` (see module docstring).
+
+    Returns {'csr_pem', 'private_key_pem', 'sans', 'subject'}.
+    """
+    key = _KEY_FACTORIES[payload.key_algorithm]()
+
+    attrs = [x509.NameAttribute(NameOID.COMMON_NAME, payload.common_name)]
+    subject_json: Dict[str, str] = {}
+    for attr_name, oid, json_key in _SUBJECT_OID_MAP:
+        value = getattr(payload, attr_name, None)
+        if value and str(value).strip():
+            cleaned = str(value).strip()
+            attrs.append(x509.NameAttribute(oid, cleaned))
+            subject_json[json_key] = cleaned
+
+    # CN always first in the SAN list, then the extra names, deduped with
+    # order preserved (mirrors the ACME flow where domains[0] is the CN).
+    sans = list(dict.fromkeys([payload.common_name, *(payload.sans or [])]))
+
+    builder = (
+        x509.CertificateSigningRequestBuilder()
+        .subject_name(x509.Name(attrs))
+        .add_extension(
+            x509.SubjectAlternativeName([x509.DNSName(d) for d in sans]),
+            critical=False,
+        )
+    )
+    csr = builder.sign(key, hashes.SHA256())
+
+    return {
+        'csr_pem': csr.public_bytes(serialization.Encoding.PEM).decode('utf-8'),
+        'private_key_pem': key.private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.NoEncryption(),
+        ).decode('utf-8'),
+        'sans': sans,
+        'subject': subject_json,
+    }
+
+
+def diff_domains(csr_sans: Optional[List[str]], cert_domains: Optional[List[str]]) -> List[str]:
+    """Human-readable warnings for SAN drift between the CSR and the signed
+    certificate (case-insensitive set diff). CAs legitimately add/normalise
+    SANs, so drift is WARN-only — the hard gate is the key match."""
+    csr_set = {d.lower() for d in (csr_sans or []) if d}
+    cert_set = {d.lower() for d in (cert_domains or []) if d}
+    warnings: List[str] = []
+    added = sorted(cert_set - csr_set)
+    dropped = sorted(csr_set - cert_set)
+    if added:
+        warnings.append(
+            f"The CA added domains that were not in the CSR: {', '.join(added)}"
+        )
+    if dropped:
+        warnings.append(
+            f"The CA dropped domains that were requested in the CSR: {', '.join(dropped)}"
+        )
+    return warnings
+
+
+def _maybe_json_list(value: Any) -> List[str]:
+    """asyncpg returns JSONB columns as str unless a codec is registered."""
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+            return parsed if isinstance(parsed, list) else []
+        except Exception:
+            return []
+    return list(value) if value else []
+
+
+def csr_row_to_dict(row: Any, include_pem: bool = False) -> Dict[str, Any]:
+    """Row → API dict. ALWAYS strips private_key_pem — the key never leaves
+    the server via a CSR endpoint. csr_pem included only on demand
+    (detail/create responses, not lists)."""
+    d = dict(row)
+    d.pop('private_key_pem', None)
+    if not include_pem:
+        d.pop('csr_pem', None)
+    for key in ('subject', 'sans'):
+        if key in d and isinstance(d[key], str):
+            try:
+                d[key] = json.loads(d[key])
+            except Exception:
+                pass
+    return d
+
+
+async def assert_csr_name_available(conn, name: str) -> None:
+    """Reject a CSR name that is already taken by an ACTIVE certificate or
+    another PENDING CSR. Called BEFORE key generation (cheap fail-fast) and
+    re-run inside `insert_csr_row` (the unique index closes the race)."""
+    existing_cert = await conn.fetchval(
+        "SELECT id FROM ssl_certificates WHERE name = $1 AND is_active = TRUE",
+        name,
+    )
+    if existing_cert:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"An active SSL certificate named '{name}' already exists. "
+                "The CSR name becomes the certificate name at import — choose "
+                "a different name or remove the existing certificate first."
+            ),
+        )
+    existing_csr = await conn.fetchval(
+        "SELECT id FROM ssl_csrs WHERE name = $1 AND status = 'pending'",
+        name,
+    )
+    if existing_csr:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"A pending CSR named '{name}' already exists (id={existing_csr}). "
+                "Import or delete it first, or choose a different name."
+            ),
+        )
+
+
+async def insert_csr_row(conn, payload: Any, bundle: Dict[str, Any], user_id: Optional[int]) -> int:
+    """Persist a freshly generated CSR bundle. Returns the new csr id."""
+    await assert_csr_name_available(conn, payload.name)
+    try:
+        csr_id = await conn.fetchval(
+            """
+            INSERT INTO ssl_csrs
+            (name, common_name, subject, sans, key_algorithm, csr_pem,
+             private_key_pem, status, created_by)
+            VALUES ($1, $2, $3::jsonb, $4::jsonb, $5, $6, $7, 'pending', $8)
+            RETURNING id
+            """,
+            payload.name,
+            payload.common_name,
+            json.dumps(bundle['subject']),
+            json.dumps(bundle['sans']),
+            payload.key_algorithm,
+            bundle['csr_pem'],
+            bundle['private_key_pem'],
+            user_id,
+        )
+    except asyncpg.exceptions.UniqueViolationError:
+        # uq_ssl_csrs_name_pending — a concurrent request won the name.
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"A pending CSR named '{payload.name}' was just created by a "
+                "concurrent request — choose a different name."
+            ),
+        )
+    return csr_id
+
+
+async def import_signed_certificate(conn, csr_id: int, imp: Any, user_id: Optional[int]) -> Dict[str, Any]:
+    """Pair the CA-signed certificate with the stored CSR key and create the
+    ssl_certificates row. Atomic: cert row + CSR state change commit together.
+
+    Returns {'certificate_id', 'certificate_name', 'primary_domain',
+    'warnings', 'reactivated'}. Raises HTTPException on every failure
+    (404 missing, 409 already completed, 400 validation).
+    """
+    async with conn.transaction():
+        # Row lock serialises concurrent imports AND a concurrent DELETE of
+        # the same CSR; works across multiple uvicorn workers (DB-level lock).
+        row = await conn.fetchrow(
+            "SELECT * FROM ssl_csrs WHERE id = $1 FOR UPDATE", csr_id
+        )
+        if not row:
+            raise HTTPException(status_code=404, detail="CSR not found")
+        if row['status'] == 'completed':
+            raise HTTPException(
+                status_code=409,
+                detail=(
+                    f"CSR '{row['name']}' is already completed — certificate "
+                    f"id {row['ssl_certificate_id']} was imported from it. "
+                    "Create a new CSR to reissue."
+                ),
+            )
+        stored_key = row['private_key_pem']
+        if not stored_key:
+            raise HTTPException(
+                status_code=500,
+                detail=(
+                    "Stored CSR private key is missing — the CSR row is "
+                    "corrupt. Delete it and create a new CSR."
+                ),
+            )
+
+        effective_name = getattr(imp, 'name', None) or row['name']
+
+        # Parse the pasted certificate FIRST so a malformed/truncated CA
+        # response gets the manual flow's 400, not a 500 from the key-match
+        # step below (verify_certificate_key_match reports an unparseable
+        # cert as match=None, which we treat as an integrity failure).
+        from utils.ssl_parser import parse_ssl_certificate, verify_certificate_key_match
+        precheck = parse_ssl_certificate(imp.certificate_content)
+        if precheck.get('error'):
+            raise HTTPException(
+                status_code=400,
+                detail=f"Invalid SSL certificate: {precheck['error']}",
+            )
+
+        # THE defining check of this feature: the CA response must match the
+        # key we generated. Deliberately stricter than create_cert_row's
+        # lenient fallback — we generated this key ourselves, so an
+        # unverifiable pair is an integrity failure, not operator input.
+        match_result = verify_certificate_key_match(imp.certificate_content, stored_key)
+        if match_result.get('match') is False:
+            raise HTTPException(
+                status_code=400,
+                detail=(
+                    "The signed certificate does not match this CSR's private "
+                    "key — the CA response likely belongs to a different "
+                    "CSR/key. Verify you pasted the certificate that was "
+                    "issued for this exact CSR."
+                ),
+            )
+        if match_result.get('match') is not True:
+            raise HTTPException(
+                status_code=500,
+                detail=(
+                    "Could not verify the certificate/key pair: "
+                    f"{match_result.get('reason', 'unknown')}"
+                ),
+            )
+
+        # Full parse/validation pipeline shared with the manual + wizard
+        # flows: invalid PEM, bad chain and already-expired certs all 400.
+        payload = SimpleNamespace(
+            name=effective_name,
+            certificate_content=imp.certificate_content,
+            private_key_content=stored_key,
+            chain_content=getattr(imp, 'chain_content', None),
+            usage_type=getattr(imp, 'usage_type', 'frontend') or 'frontend',
+        )
+        fields = ssl_service._prepare_cert_fields(payload)
+
+        # Global name uniqueness (ssl_certificates.cluster_id is always NULL
+        # under the R38 schema, so name is effectively a global namespace).
+        existing = await conn.fetchrow(
+            "SELECT id, is_active FROM ssl_certificates WHERE name = $1 LIMIT 1",
+            effective_name,
+        )
+        if existing and existing['is_active']:
+            raise HTTPException(
+                status_code=400,
+                detail=(
+                    f"An active SSL certificate named '{effective_name}' "
+                    "already exists (created after this CSR). Delete or "
+                    "rename it, or pass a different `name` in the import "
+                    "request — the CSR stays pending and can be re-imported."
+                ),
+            )
+
+        reactivated = False
+        if existing and not existing['is_active']:
+            # Reactivate the soft-deleted row (mirrors create_cert_row):
+            # preserves the row id so historical references keep working.
+            await conn.execute(
+                "DELETE FROM ssl_certificate_clusters WHERE ssl_certificate_id = $1",
+                existing['id'],
+            )
+            await conn.execute(
+                """
+                UPDATE ssl_certificates
+                SET is_active = TRUE,
+                    last_config_status = 'PENDING',
+                    certificate_content = $2,
+                    private_key_content = $3,
+                    chain_content = $4,
+                    primary_domain = $5,
+                    all_domains = $6::jsonb,
+                    expiry_date = $7,
+                    usage_type = $8,
+                    issuer = $9,
+                    fingerprint = $10,
+                    status = $11,
+                    days_until_expiry = $12,
+                    source = 'csr',
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE id = $1
+                """,
+                existing['id'],
+                fields['cert_content'],
+                fields['private_key_content'],
+                fields['chain_content'],
+                fields['primary_domain'],
+                json.dumps(fields['all_domains']),
+                fields['expiry_date'],
+                fields['usage_type'],
+                fields['issuer'],
+                fields['fingerprint'],
+                fields['status'],
+                fields['days_until_expiry'],
+            )
+            cert_id = existing['id']
+            reactivated = True
+            logger.info(
+                f"csr_service.import_signed_certificate: reactivated "
+                f"soft-deleted cert '{effective_name}' (id={cert_id}) for CSR {csr_id}"
+            )
+        else:
+            cert_id = await conn.fetchval(
+                """
+                INSERT INTO ssl_certificates (
+                    name, primary_domain, certificate_content, private_key_content,
+                    chain_content, expiry_date, issuer, fingerprint, status,
+                    days_until_expiry, all_domains, is_active, cluster_id,
+                    last_config_status, usage_type, source
+                ) VALUES (
+                    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb,
+                    TRUE, NULL, 'PENDING', $12, 'csr'
+                )
+                RETURNING id
+                """,
+                effective_name,
+                fields['primary_domain'],
+                fields['cert_content'],
+                fields['private_key_content'],
+                fields['chain_content'],
+                fields['expiry_date'],
+                fields['issuer'],
+                fields['fingerprint'],
+                fields['status'],
+                fields['days_until_expiry'],
+                json.dumps(fields['all_domains']),
+                fields['usage_type'],
+            )
+
+        # Cluster bindings: global = zero junction rows (existing convention).
+        if not getattr(imp, 'is_global', False):
+            for cluster_id in (getattr(imp, 'cluster_ids', None) or []):
+                await ssl_service.ensure_cluster_junction(conn, cert_id, cluster_id)
+
+        # Complete the CSR and destroy the key copy — the key now lives on
+        # the certificate row only, like every other key in the system.
+        await conn.execute(
+            """
+            UPDATE ssl_csrs
+            SET status = 'completed',
+                ssl_certificate_id = $2,
+                private_key_pem = NULL,
+                completed_at = CURRENT_TIMESTAMP,
+                updated_at = CURRENT_TIMESTAMP
+            WHERE id = $1
+            """,
+            csr_id,
+            cert_id,
+        )
+
+    warnings = diff_domains(_maybe_json_list(row['sans']), fields['all_domains'])
+    if reactivated:
+        warnings.append(
+            f"A soft-deleted certificate named '{effective_name}' was "
+            f"reactivated (row id {cert_id}) — existing entities that still "
+            "reference that id now serve the newly imported certificate."
+        )
+
+    return {
+        'certificate_id': cert_id,
+        'certificate_name': effective_name,
+        'primary_domain': fields['primary_domain'],
+        'warnings': warnings,
+        'reactivated': reactivated,
+    }

backend/services/ssl_service.py

@@ -40,10 +40,12 @@ flow.
   (callers translate to wizard step-jumpback toasts).
 """
 
+import hashlib
 import json
 import logging
+import time
 from datetime import datetime, timezone
-from typing import Any, Optional
+from typing import Any, List, Optional
 
 from fastapi import HTTPException
 
@@ -106,23 +108,22 @@ def _recompute_status_from_expiry(
         return cert_info_status or "valid", cert_info_days or 0
 
 
-async def create_cert_row(
-    conn,
-    payload: Any,
-    cluster_id: int,
-) -> int:
-    """Insert a row into ssl_certificates (always cluster_id=NULL) + junction
-    binding to the given cluster_id. Returns new ssl_certificate_id.
+def _prepare_cert_fields(payload: Any) -> dict:
+    """Parse + validate the PEM material on `payload` and derive every
+    ssl_certificates column value from it (v1.9.0 extraction — shared by
+    `create_cert_row` and the CSR import flow in services/csr_service.py,
+    byte-identical to the former inline body of `create_cert_row`).
 
     payload is expected to expose:
       name, certificate_content, private_key_content, chain_content,
       usage_type (optional, default 'frontend').
 
-    All cert metadata (primary_domain, all_domains, expiry_date,
-    issuer, fingerprint, status, days_until_expiry) is now parsed
-    FROM the PEM content via `parse_ssl_certificate` — operator-
-    supplied values on the payload are accepted as a graceful
-    fallback only when parsing fails (which itself raises 400).
+    Raises HTTPException(400) on any parse/validation failure (invalid PEM,
+    bad private key, cert/key mismatch, bad chain, already-expired cert).
+
+    Returns a dict with keys: cert_content, private_key_content,
+    chain_content, cert_info, primary_domain, all_domains, expiry_date,
+    issuer, fingerprint, status, days_until_expiry, usage_type.
     """
     cert_content = getattr(payload, "certificate_content", None) or ""
     if not cert_content.strip():
@@ -213,6 +214,53 @@ async def create_cert_row(
     )
     usage_type = getattr(payload, "usage_type", "frontend") or "frontend"
 
+    return {
+        "cert_content": cert_content,
+        "private_key_content": private_key_content,
+        "chain_content": chain_content,
+        "cert_info": cert_info,
+        "primary_domain": primary_domain,
+        "all_domains": all_domains,
+        "expiry_date": expiry_date,
+        "issuer": issuer,
+        "fingerprint": fingerprint,
+        "status": status,
+        "days_until_expiry": days_until_expiry,
+        "usage_type": usage_type,
+    }
+
+
+async def create_cert_row(
+    conn,
+    payload: Any,
+    cluster_id: int,
+) -> int:
+    """Insert a row into ssl_certificates (always cluster_id=NULL) + junction
+    binding to the given cluster_id. Returns new ssl_certificate_id.
+
+    payload is expected to expose:
+      name, certificate_content, private_key_content, chain_content,
+      usage_type (optional, default 'frontend').
+
+    All cert metadata (primary_domain, all_domains, expiry_date,
+    issuer, fingerprint, status, days_until_expiry) is now parsed
+    FROM the PEM content via `parse_ssl_certificate` — operator-
+    supplied values on the payload are accepted as a graceful
+    fallback only when parsing fails (which itself raises 400).
+    """
+    fields = _prepare_cert_fields(payload)
+    cert_content = fields["cert_content"]
+    private_key_content = fields["private_key_content"]
+    chain_content = fields["chain_content"]
+    expiry_date = fields["expiry_date"]
+    primary_domain = fields["primary_domain"]
+    all_domains = fields["all_domains"]
+    issuer = fields["issuer"]
+    fingerprint = fields["fingerprint"]
+    status = fields["status"]
+    days_until_expiry = fields["days_until_expiry"]
+    usage_type = fields["usage_type"]
+
     existing = await conn.fetchrow(
         """
         SELECT s.id, s.is_active
@@ -408,3 +456,81 @@ async def validate_server_ca_bundle_eligibility(
         cluster_id,
     )
     return row is not None
+
+
+async def stage_ssl_config_versions(
+    conn,
+    cert_id: int,
+    cluster_ids: List[int],
+    action: str = "create",
+    created_by: Optional[int] = None,
+) -> List[dict]:
+    """Stage one PENDING config version per affected cluster after an SSL
+    certificate mutation (v1.9.0 — distilled from the routers/ssl.py POST
+    /certificates staging loop; used by the CSR import flow).
+
+    Uses the EXACT `ssl-{cert_id}-{action}-{timestamp}` version-name scheme of
+    the manual SSL flow so Apply Management, the `has_pending_config`
+    LIKE-filter ('ssl-' || id || '-%'), and the agent delivery predicates
+    treat CSR-imported certificates identically to manually uploaded ones.
+    Agents are NOT notified here — the operator applies manually.
+
+    Per-cluster failures are caught and reported in the returned
+    sync_results list (the DB save has already succeeded — same semantics as
+    the manual flow, where a config-generation failure never rolls back the
+    certificate row).
+    """
+    # Local import: keeps services/haproxy_config free to import ssl helpers
+    # without a module-level cycle.
+    from services.haproxy_config import generate_haproxy_config_for_cluster
+
+    sync_results: List[dict] = []
+    for cluster_id in cluster_ids:
+        try:
+            config_content = await generate_haproxy_config_for_cluster(cluster_id)
+            config_hash = hashlib.sha256(config_content.encode()).hexdigest()
+            version_name = f"ssl-{cert_id}-{action}-{int(time.time())}"
+
+            version_created_by = created_by
+            if version_created_by is None:
+                version_created_by = await conn.fetchval(
+                    "SELECT id FROM users WHERE username = 'admin' LIMIT 1"
+                ) or 1
+
+            await conn.fetchval(
+                """
+                INSERT INTO config_versions
+                (cluster_id, version_name, config_content, checksum, created_by, is_active, status)
+                VALUES ($1, $2, $3, $4, $5, FALSE, 'PENDING')
+                RETURNING id
+                """,
+                cluster_id,
+                version_name,
+                config_content,
+                config_hash,
+                version_created_by,
+            )
+            logger.info(
+                f"APPLY WORKFLOW: Created PENDING config version {version_name} "
+                f"for cluster {cluster_id} (ssl_service.stage_ssl_config_versions)"
+            )
+            sync_results.append({
+                'node': 'pending',
+                'success': True,
+                'cluster_id': cluster_id,
+                'version': version_name,
+                'status': 'PENDING',
+                'message': 'SSL certificate staged. Click Apply to activate.',
+            })
+        except Exception as e:
+            logger.error(
+                f"Cluster config staging failed for SSL certificate {cert_id} "
+                f"on cluster {cluster_id}: {e}"
+            )
+            sync_results.append({
+                'node': 'cluster',
+                'success': False,
+                'cluster_id': cluster_id,
+                'error': str(e),
+            })
+    return sync_results

backend/tests/test_csr_import.py

@@ -0,0 +1,469 @@
+"""
+v1.9.0 CSR creation — unit tests for the signed-certificate import flow and
+config-version staging (pattern: test_ssl_service_extraction.py, AsyncMock conn).
+
+Pins the security-relevant invariants:
+- key match is a HARD gate: match=False → 400 before any INSERT, and
+  match=None (unverifiable) → 500, never a lenient pass (we generated the
+  key ourselves — deliberate divergence from create_cert_row's fallback).
+- the new cert row is cluster_id=NULL / last_config_status='PENDING' /
+  source='csr' (PENDING keeps it invisible to agents until Apply).
+- completing the CSR NULLs the private key copy.
+- staging reuses the exact `ssl-{id}-create-{ts}` version-name scheme.
+"""
+import json
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from models.csr import SSLCSRImport
+from services.csr_service import (
+    assert_csr_name_available,
+    import_signed_certificate,
+    insert_csr_row,
+)
+from services.ssl_service import stage_ssl_config_versions
+
+
+_VALID_PARSE = {
+    "primary_domain": "www.example.com",
+    "all_domains": ["www.example.com"],
+    "expiry_date": datetime(2099, 1, 1, tzinfo=timezone.utc),
+    "issuer": "CN=Test CA",
+    "fingerprint": "AA:BB:CC",
+    "status": "valid",
+    "days_until_expiry": 365,
+}
+
+_FAKE_CERT = "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----"
+_FAKE_KEY = "-----BEGIN PRIVATE KEY-----\nY\n-----END PRIVATE KEY-----"
+
+
+def _csr_row(**overrides):
+    row = {
+        "id": 5,
+        "name": "csr-www",
+        "common_name": "www.example.com",
+        "subject": "{}",
+        "sans": json.dumps(["www.example.com"]),
+        "key_algorithm": "rsa-2048",
+        "csr_pem": "-----BEGIN CERTIFICATE REQUEST-----\nZ\n-----END CERTIFICATE REQUEST-----",
+        "private_key_pem": _FAKE_KEY,
+        "status": "pending",
+        "ssl_certificate_id": None,
+    }
+    row.update(overrides)
+    return row
+
+
+def _mk_conn():
+    conn = AsyncMock()
+    # asyncpg's conn.transaction() is a SYNC call returning an async CM.
+    conn.transaction = MagicMock()
+    return conn
+
+
+def _import_payload(**overrides):
+    base = dict(
+        certificate_content=_FAKE_CERT,
+        chain_content=None,
+        usage_type="frontend",
+        is_global=False,
+        cluster_ids=[1, 2],
+        name=None,
+    )
+    base.update(overrides)
+    return SSLCSRImport(**base)
+
+
+@contextmanager
+def _patched(match=None, parse=None):
+    """Patch every parser touchpoint of the import path: the function-local
+    imports in csr_service (utils.ssl_parser.*) and the module-level imports
+    in ssl_service._prepare_cert_fields (services.ssl_service.*)."""
+    match_result = match if match is not None else {"match": True}
+    parse_result = dict(parse or _VALID_PARSE)
+    with patch("utils.ssl_parser.verify_certificate_key_match", return_value=match_result), \
+         patch("utils.ssl_parser.parse_ssl_certificate", return_value=dict(parse_result)), \
+         patch("services.ssl_service.parse_ssl_certificate", return_value=dict(parse_result)), \
+         patch("services.ssl_service.validate_private_key", return_value=True), \
+         patch("services.ssl_service.validate_certificate_chain", return_value=True):
+        yield
+
+
+# ----------------------------------------------------------------------------
+# import_signed_certificate
+# ----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_import_happy_path_inserts_pending_csr_sourced_cert():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row(), None]  # FOR UPDATE row, no name clash
+    conn.fetchval.return_value = 42                 # INSERT ... RETURNING id
+
+    with _patched():
+        result = await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert result["certificate_id"] == 42
+    assert result["reactivated"] is False
+
+    # Concurrency invariants: everything runs inside a transaction and the
+    # CSR row is locked FOR UPDATE (serialises double-import and delete-races).
+    assert conn.transaction.call_count == 1
+    lock_sql = conn.fetchrow.call_args_list[0].args[0]
+    assert "FOR UPDATE" in lock_sql
+
+    insert_sql, *insert_args = conn.fetchval.call_args.args
+    assert "INSERT INTO ssl_certificates" in insert_sql
+    assert "NULL, 'PENDING'" in insert_sql, "cert must stay invisible to agents until Apply"
+    assert "'csr'" in insert_sql, "source column must record the CSR origin"
+    # The stored CSR key — not any request-supplied key — must be persisted.
+    assert _FAKE_KEY in insert_args
+
+    # One junction row per requested cluster.
+    junction_calls = [
+        c for c in conn.execute.call_args_list
+        if c.args and "ssl_certificate_clusters" in c.args[0] and "INSERT" in c.args[0]
+    ]
+    assert len(junction_calls) == 2
+    assert {c.args[2] for c in junction_calls} == {1, 2}
+
+    # CSR completion must destroy the key copy.
+    completion_calls = [
+        c for c in conn.execute.call_args_list
+        if c.args and "UPDATE ssl_csrs" in c.args[0]
+    ]
+    assert len(completion_calls) == 1
+    assert "private_key_pem = NULL" in completion_calls[0].args[0]
+    assert "status = 'completed'" in completion_calls[0].args[0]
+    assert completion_calls[0].args[1] == 5   # csr_id
+    assert completion_calls[0].args[2] == 42  # cert_id
+
+
+@pytest.mark.asyncio
+async def test_import_global_creates_zero_junction_rows():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row(), None]
+    conn.fetchval.return_value = 42
+
+    with _patched():
+        await import_signed_certificate(
+            conn, 5, _import_payload(is_global=True, cluster_ids=None), user_id=7
+        )
+
+    junction_calls = [
+        c for c in conn.execute.call_args_list
+        if c.args and "ssl_certificate_clusters" in c.args[0] and "INSERT" in c.args[0]
+    ]
+    assert junction_calls == [], "global cert = zero junction rows (existing convention)"
+
+
+@pytest.mark.asyncio
+async def test_import_key_mismatch_rejected_400_before_any_write():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row()]
+
+    with _patched(match={"match": False, "reason": "public key mismatch"}):
+        with pytest.raises(HTTPException) as exc_info:
+            await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert exc_info.value.status_code == 400
+    assert "does not match" in exc_info.value.detail
+    assert not conn.fetchval.await_count, "nothing must be inserted on mismatch"
+    assert not conn.execute.await_count
+
+
+@pytest.mark.asyncio
+async def test_import_unverifiable_key_match_is_hard_error_not_lenient():
+    """match=None means OUR stored key is unreadable — integrity failure,
+    never the lenient pass create_cert_row historically allows."""
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row()]
+
+    with _patched(match={"match": None, "reason": "key could not be parsed"}):
+        with pytest.raises(HTTPException) as exc_info:
+            await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert exc_info.value.status_code == 500
+    assert not conn.fetchval.await_count
+
+
+@pytest.mark.asyncio
+async def test_import_expired_certificate_rejected_400():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row()]
+
+    expired = dict(_VALID_PARSE)
+    expired["status"] = "expired"
+    expired["days_until_expiry"] = -10
+    with _patched(parse=expired):
+        with pytest.raises(HTTPException) as exc_info:
+            await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert exc_info.value.status_code == 400
+    assert "expired" in exc_info.value.detail.lower()
+    assert not conn.fetchval.await_count
+
+
+@pytest.mark.asyncio
+async def test_import_malformed_certificate_rejected_400_not_500():
+    """A cert with PEM markers but unparseable content (truncated CA response)
+    is OPERATOR INPUT — it must get the manual flow's 400, not the 500 that
+    the strict key-match branch reserves for a corrupt STORED key."""
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row()]
+
+    with _patched(parse={"error": "Could not parse certificate"}):
+        with pytest.raises(HTTPException) as exc_info:
+            await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert exc_info.value.status_code == 400
+    assert "Invalid SSL certificate" in exc_info.value.detail
+    assert not conn.fetchval.await_count
+    assert not conn.execute.await_count
+
+
+@pytest.mark.asyncio
+async def test_import_completed_csr_conflicts_409():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row(status="completed", ssl_certificate_id=42)]
+
+    with pytest.raises(HTTPException) as exc_info:
+        await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert exc_info.value.status_code == 409
+    assert "already completed" in exc_info.value.detail
+
+
+@pytest.mark.asyncio
+async def test_import_missing_csr_404():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [None]
+
+    with pytest.raises(HTTPException) as exc_info:
+        await import_signed_certificate(conn, 999, _import_payload(), user_id=7)
+
+    assert exc_info.value.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_import_active_name_collision_rejected_with_hint():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row(), {"id": 9, "is_active": True}]
+
+    with _patched():
+        with pytest.raises(HTTPException) as exc_info:
+            await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert exc_info.value.status_code == 400
+    assert "already exists" in exc_info.value.detail
+    assert "name" in exc_info.value.detail  # points at the override escape hatch
+    assert not conn.fetchval.await_count
+
+
+@pytest.mark.asyncio
+async def test_import_name_override_is_used_for_the_cert_row():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row(), None]
+    conn.fetchval.return_value = 42
+
+    with _patched():
+        result = await import_signed_certificate(
+            conn, 5, _import_payload(name="renamed-cert"), user_id=7
+        )
+
+    assert result["certificate_name"] == "renamed-cert"
+    _, *insert_args = conn.fetchval.call_args.args
+    assert "renamed-cert" in insert_args
+    # And the collision check must have run against the override, not csr.name.
+    name_lookup = conn.fetchrow.call_args_list[1]
+    assert name_lookup.args[1] == "renamed-cert"
+
+
+@pytest.mark.asyncio
+async def test_import_reactivates_soft_deleted_name_and_warns():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [_csr_row(), {"id": 77, "is_active": False}]
+
+    with _patched():
+        result = await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert result["certificate_id"] == 77
+    assert result["reactivated"] is True
+    assert any("reactivated" in w for w in result["warnings"])
+    assert not conn.fetchval.await_count, "reactivation must UPDATE, not INSERT"
+    update_calls = [
+        c for c in conn.execute.call_args_list
+        if c.args and "UPDATE ssl_certificates" in c.args[0]
+    ]
+    assert len(update_calls) == 1
+    update_sql = update_calls[0].args[0]
+    assert "source = 'csr'" in update_sql
+    # The reactivated row must come back to life invisible to agents until
+    # Apply, with the row itself active again.
+    assert "last_config_status = 'PENDING'" in update_sql
+    assert "is_active = TRUE" in update_sql
+    # Old cluster bindings must be wiped before re-binding to the new scope.
+    junction_deletes = [
+        c for c in conn.execute.call_args_list
+        if c.args and "DELETE FROM ssl_certificate_clusters" in c.args[0]
+    ]
+    assert len(junction_deletes) == 1
+    assert junction_deletes[0].args[1] == 77
+    # …and the importer's requested clusters re-bound via the junction.
+    junction_inserts = [
+        c for c in conn.execute.call_args_list
+        if c.args and "INSERT INTO ssl_certificate_clusters" in c.args[0]
+    ]
+    assert {c.args[2] for c in junction_inserts} == {1, 2}
+
+
+@pytest.mark.asyncio
+async def test_import_san_drift_warns_but_succeeds():
+    conn = _mk_conn()
+    conn.fetchrow.side_effect = [
+        _csr_row(sans=json.dumps(["www.example.com", "api.example.com"])),
+        None,
+    ]
+    conn.fetchval.return_value = 42
+
+    drifted = dict(_VALID_PARSE)
+    drifted["all_domains"] = ["www.example.com", "cdn.example.com"]
+    with _patched(parse=drifted):
+        result = await import_signed_certificate(conn, 5, _import_payload(), user_id=7)
+
+    assert result["certificate_id"] == 42
+    assert any("added" in w and "cdn.example.com" in w for w in result["warnings"])
+    assert any("dropped" in w and "api.example.com" in w for w in result["warnings"])
+
+
+# ----------------------------------------------------------------------------
+# insert_csr_row / assert_csr_name_available
+# ----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_csr_name_taken_by_active_cert_rejected():
+    conn = _mk_conn()
+    conn.fetchval.side_effect = [11]  # active cert with the name exists
+
+    with pytest.raises(HTTPException) as exc_info:
+        await assert_csr_name_available(conn, "taken")
+    assert exc_info.value.status_code == 400
+    assert "certificate" in exc_info.value.detail.lower()
+
+
+@pytest.mark.asyncio
+async def test_csr_name_taken_by_pending_csr_rejected():
+    conn = _mk_conn()
+    conn.fetchval.side_effect = [None, 12]  # no cert, but a pending CSR
+
+    with pytest.raises(HTTPException) as exc_info:
+        await assert_csr_name_available(conn, "taken")
+    assert exc_info.value.status_code == 400
+    assert "pending CSR" in exc_info.value.detail
+
+
+@pytest.mark.asyncio
+async def test_insert_csr_row_translates_unique_violation_to_400():
+    """The uq_ssl_csrs_name_pending partial index closes the create/create
+    race — the loser must get a clean 400, not a 500."""
+    import asyncpg as _asyncpg
+
+    conn = _mk_conn()
+    # availability checks pass, INSERT hits the unique index
+    conn.fetchval.side_effect = [
+        None, None, _asyncpg.exceptions.UniqueViolationError("dup"),
+    ]
+    payload = SimpleNamespace(
+        name="raced", common_name="www.example.com", key_algorithm="rsa-2048"
+    )
+    bundle = {"subject": {}, "sans": ["www.example.com"], "csr_pem": "PEM", "private_key_pem": "KEY"}
+
+    with pytest.raises(HTTPException) as exc_info:
+        await insert_csr_row(conn, payload, bundle, user_id=1)
+    assert exc_info.value.status_code == 400
+    assert "concurrent" in exc_info.value.detail
+
+
+# ----------------------------------------------------------------------------
+# router-level guards
+# ----------------------------------------------------------------------------
+
+
+def test_cluster_id_int32_guard_rejects_out_of_range_with_404():
+    """Body-supplied cluster ids must never reach asyncpg out of int4 range
+    (DataError → raw 500) — same Bulgu #96 hygiene as the csr_id path param."""
+    from routers.csr import _assert_valid_cluster_id
+
+    _assert_valid_cluster_id(1)
+    _assert_valid_cluster_id(2_147_483_647)
+    for bad in (0, -1, 2_147_483_648, 99_999_999_999):
+        with pytest.raises(HTTPException) as exc_info:
+            _assert_valid_cluster_id(bad)
+        assert exc_info.value.status_code == 404
+        assert "Cluster not found" in exc_info.value.detail
+
+
+# ----------------------------------------------------------------------------
+# stage_ssl_config_versions
+# ----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_stage_creates_one_pending_version_per_cluster_with_ssl_naming():
+    import re
+
+    conn = _mk_conn()
+    conn.fetchval.return_value = 1001  # config_versions INSERT RETURNING id
+
+    with patch(
+        "services.haproxy_config.generate_haproxy_config_for_cluster",
+        new=AsyncMock(return_value="# cfg"),
+    ):
+        results = await stage_ssl_config_versions(conn, 42, [1, 2], created_by=7)
+
+    assert len(results) == 2
+    assert all(r["success"] for r in results)
+    assert [r["cluster_id"] for r in results] == [1, 2]
+
+    insert_calls = [
+        c for c in conn.fetchval.call_args_list
+        if c.args and "INSERT INTO config_versions" in c.args[0]
+    ]
+    assert len(insert_calls) == 2
+    for call in insert_calls:
+        sql = call.args[0]
+        assert "FALSE, 'PENDING'" in sql, "staged versions must be inactive + PENDING"
+        version_name = call.args[2]
+        # EXACT manual-flow scheme: Apply Management + has_pending_config
+        # LIKE-filters key off 'ssl-{id}-...'.
+        assert re.match(r"^ssl-42-create-\d+$", version_name), version_name
+        assert call.args[5] == 7  # created_by honours the importing user
+
+
+@pytest.mark.asyncio
+async def test_stage_reports_per_cluster_failure_without_raising():
+    conn = _mk_conn()
+    conn.fetchval.return_value = 1001
+
+    async def _gen(cluster_id):
+        if cluster_id == 2:
+            raise RuntimeError("config generation exploded")
+        return "# cfg"
+
+    with patch(
+        "services.haproxy_config.generate_haproxy_config_for_cluster",
+        new=AsyncMock(side_effect=_gen),
+    ):
+        results = await stage_ssl_config_versions(conn, 42, [1, 2], created_by=7)
+
+    assert len(results) == 2
+    assert results[0]["success"] is True
+    assert results[1]["success"] is False
+    assert "exploded" in results[1]["error"]

backend/tests/test_csr_migration.py

@@ -0,0 +1,104 @@
+"""
+v1.9.0 CSR creation — static source assertions (pattern: test_vip_purge.py).
+
+Guards the migration wiring that a unit test cannot exercise without a real
+database: the SCHEMA_VERSION bump (without it, deployed installs skip the
+whole migration run and the ssl_csrs table never appears), the migration
+registration, the security-relevant DDL, and the router registration.
+"""
+import os
+import re
+
+_BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _read(rel_path: str) -> str:
+    with open(os.path.join(_BACKEND_DIR, rel_path), encoding="utf-8") as f:
+        return f.read()
+
+
+def test_schema_version_bumped_to_10():
+    src = _read(os.path.join("database", "migrations.py"))
+    m = re.search(r"^SCHEMA_VERSION\s*=\s*(\d+)", src, re.MULTILINE)
+    assert m, "SCHEMA_VERSION constant not found in migrations.py"
+    assert int(m.group(1)) >= 10, (
+        "SCHEMA_VERSION must be >= 10 for the v1.9.0 ssl_csrs table — "
+        "without the bump, existing installs (version >= 9) skip the whole "
+        "migration run and never gain the table."
+    )
+
+
+def test_ssl_csrs_migration_defined_and_registered():
+    src = _read(os.path.join("database", "migrations.py"))
+    assert "async def ensure_ssl_csrs_table" in src
+
+    inner = src.split("async def _run_all_migrations_inner", 1)[1]
+    inner = inner.split("\nasync def ", 1)[0]  # body of the runner only
+    assert "await ensure_ssl_csrs_table()" in inner, (
+        "ensure_ssl_csrs_table must be invoked from _run_all_migrations_inner"
+    )
+
+
+def test_ssl_csrs_ddl_essentials():
+    src = _read(os.path.join("database", "migrations.py"))
+    ddl_start = src.index("CREATE TABLE IF NOT EXISTS ssl_csrs")
+    ddl = src[ddl_start:ddl_start + 2500]
+
+    assert "private_key_pem TEXT" in ddl
+    assert "name VARCHAR(100) NOT NULL" in ddl, (
+        "ssl_csrs.name must align with ssl_certificates.name VARCHAR(100)"
+    )
+    assert "ssl_certificate_id INTEGER REFERENCES ssl_certificates(id) ON DELETE SET NULL" in ddl, (
+        "deleting the imported cert must not cascade into CSR history"
+    )
+    # Partial unique index: only PENDING CSRs reserve their target cert name.
+    assert "uq_ssl_csrs_name_pending" in src
+    assert re.search(
+        r"uq_ssl_csrs_name_pending\s+ON\s+ssl_csrs\(name\)\s+WHERE\s+status\s*=\s*'pending'",
+        src,
+    ), "name uniqueness must be scoped to pending CSRs (partial index)"
+
+
+def test_csr_router_registered_in_main():
+    src = _read("main.py")
+    assert "from routers.csr import router as csr_router" in src
+    assert "app.include_router(csr_router)" in src
+
+
+def test_csr_endpoint_permission_mapping():
+    """Pin which ssl.<action> permission each endpoint enforces: a regression
+    that dropped or weakened a _require() call would otherwise pass the
+    auth-rejection tests (they only assert 401/403 for unauthenticated calls)."""
+    src = _read(os.path.join("routers", "csr.py"))
+
+    def _handler_body(decorator):
+        start = src.index(decorator)
+        nxt = src.find("@router.", start + 1)
+        return src[start:nxt if nxt != -1 else len(src)]
+
+    expectations = [
+        ('@router.post("")', '"create"'),
+        ('@router.get("")', '"read"'),
+        ('@router.get("/{csr_id}")', '"read"'),
+        ('@router.post("/{csr_id}/import")', '"create"'),
+        ('@router.delete("/{csr_id}")', '"delete"'),
+    ]
+    for decorator, action in expectations:
+        body = _handler_body(decorator)
+        assert f"_require(authorization, {action})" in body, (
+            f"endpoint {decorator} must enforce ssl.{action.strip(chr(34))}"
+        )
+
+
+def test_csr_router_never_selects_private_key():
+    """The CSR endpoints must use the explicit column list — a bare
+    `SELECT *` into an API response is how the key would leak. The one place
+    SELECT * is allowed is the service-layer FOR UPDATE row (it needs the key
+    to pair with the cert); the router itself must not touch the column."""
+    src = _read(os.path.join("routers", "csr.py"))
+    code_only = re.sub(r"#.*", "", src)  # strip comments; the column name may
+    # legitimately appear there as documentation
+    assert "private_key_pem" not in code_only, (
+        "routers/csr.py must never reference private_key_pem in code"
+    )
+    assert "SELECT *" not in code_only, "routers/csr.py must use explicit column lists"

backend/tests/test_csr_models.py

@@ -0,0 +1,172 @@
+"""
+v1.9.0 CSR creation — Pydantic model validation tests (models/csr.py).
+
+The CSR name shares the SSL certificate name's path-traversal contract
+(Bulgu #21) with one deliberate tightening: max 100 chars, matching the
+ssl_certificates.name VARCHAR(100) column.
+"""
+import pytest
+from pydantic import ValidationError
+
+from models.csr import SSLCSRCreate, SSLCSRImport
+
+_CERT_PEM = "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----"
+
+
+def _create(**overrides):
+    base = dict(name="my-csr", common_name="www.example.com")
+    base.update(overrides)
+    return SSLCSRCreate(**base)
+
+
+# ----------------------------------------------------------------------------
+# SSLCSRCreate
+# ----------------------------------------------------------------------------
+
+
+def test_minimal_valid_create():
+    m = _create()
+    assert m.name == "my-csr"
+    assert m.common_name == "www.example.com"
+    assert m.key_algorithm == "rsa-2048"
+    assert m.sans == []
+
+
+@pytest.mark.parametrize("bad_name", [
+    "../../etc/cron.d/evil",   # path traversal
+    "a..b",                    # embedded ..
+    ".hidden",                 # hidden filename
+    "-flag",                   # CLI flag confusion
+    "has space",
+    "wild*card",
+    "",
+    "x" * 101,                 # VARCHAR(100) alignment — 200 is NOT allowed here
+])
+def test_name_rejects_unsafe_values(bad_name):
+    with pytest.raises(ValidationError):
+        _create(name=bad_name)
+
+
+def test_name_accepts_100_chars():
+    assert _create(name="x" * 100).name == "x" * 100
+
+
+def test_common_name_wildcard_accepted_and_lowercased():
+    m = _create(common_name="*.Example.COM")
+    assert m.common_name == "*.example.com"
+
+
+@pytest.mark.parametrize("bad_cn", [
+    "",
+    "under_score.example.com",      # _ is not LDH
+    "*.*.example.com",              # wildcard only as leftmost single label
+    "-leading.example.com",
+    "a" * 70 + ".example.com",      # label > 63
+    "cn-longer-than-64-chars-" + "x" * 45 + ".example.com",  # CN > 64 total
+])
+def test_common_name_rejects_invalid(bad_cn):
+    with pytest.raises(ValidationError):
+        _create(common_name=bad_cn)
+
+
+def test_sans_normalised_deduped_and_capped():
+    m = _create(sans=["API.example.com", "api.example.com", "cdn.example.com"])
+    assert m.sans == ["api.example.com", "cdn.example.com"]
+
+    with pytest.raises(ValidationError):
+        _create(sans=[f"h{i}.example.com" for i in range(101)])
+
+
+def test_country_normalised_or_rejected():
+    assert _create(country="tr").country == "TR"
+    assert _create(country=None).country is None
+    for bad in ("TUR", "T", "1A"):
+        with pytest.raises(ValidationError):
+            _create(country=bad)
+
+
+def test_subject_fields_reject_control_characters():
+    with pytest.raises(ValidationError):
+        _create(organization="Evil\x00Corp")
+    with pytest.raises(ValidationError):
+        _create(locality="line\nbreak")
+
+
+def test_subject_fields_reject_overlength():
+    with pytest.raises(ValidationError):
+        _create(organization="x" * 65)
+
+
+def test_key_algorithm_strict_enum():
+    for good in ("rsa-2048", "rsa-4096", "ecdsa-p256", "ecdsa-p384"):
+        assert _create(key_algorithm=good).key_algorithm == good
+    for bad in ("rsa-1024", "rsa-8192", "ed25519", "2048", ""):
+        with pytest.raises(ValidationError):
+            _create(key_algorithm=bad)
+
+
+def test_email_basic_validation():
+    assert _create(email="ops@example.com").email == "ops@example.com"
+    with pytest.raises(ValidationError):
+        _create(email="not-an-email")
+
+
+# ----------------------------------------------------------------------------
+# SSLCSRImport
+# ----------------------------------------------------------------------------
+
+
+def test_import_minimal_global():
+    m = SSLCSRImport(certificate_content=_CERT_PEM, is_global=True)
+    assert m.usage_type == "frontend"
+    assert m.name is None
+
+
+def test_import_requires_clusters_when_not_global():
+    with pytest.raises(ValidationError):
+        SSLCSRImport(certificate_content=_CERT_PEM, is_global=False)
+    with pytest.raises(ValidationError):
+        SSLCSRImport(certificate_content=_CERT_PEM, is_global=False, cluster_ids=[])
+    m = SSLCSRImport(certificate_content=_CERT_PEM, is_global=False, cluster_ids=[1])
+    assert m.cluster_ids == [1]
+
+
+def test_import_certificate_must_be_pem():
+    with pytest.raises(ValidationError):
+        SSLCSRImport(certificate_content="not a pem", is_global=True)
+    with pytest.raises(ValidationError):
+        SSLCSRImport(certificate_content="", is_global=True)
+
+
+def test_import_certificate_size_capped():
+    huge = _CERT_PEM + "A" * (64 * 1024 + 1)
+    with pytest.raises(ValidationError):
+        SSLCSRImport(certificate_content=huge, is_global=True)
+
+
+def test_import_chain_optional_but_validated():
+    m = SSLCSRImport(certificate_content=_CERT_PEM, is_global=True, chain_content="  ")
+    assert m.chain_content is None
+    with pytest.raises(ValidationError):
+        SSLCSRImport(
+            certificate_content=_CERT_PEM, is_global=True, chain_content="garbage"
+        )
+
+
+def test_import_name_override_shares_the_name_contract():
+    m = SSLCSRImport(certificate_content=_CERT_PEM, is_global=True, name="renamed")
+    assert m.name == "renamed"
+    with pytest.raises(ValidationError):
+        SSLCSRImport(certificate_content=_CERT_PEM, is_global=True, name="../evil")
+    # Empty override collapses to None (falls back to the CSR's own name).
+    m2 = SSLCSRImport(certificate_content=_CERT_PEM, is_global=True, name="  ")
+    assert m2.name is None
+
+
+def test_import_usage_type_enum():
+    for good in ("frontend", "server"):
+        assert SSLCSRImport(
+            certificate_content=_CERT_PEM, is_global=True, usage_type=good
+        ).usage_type == good
+    with pytest.raises(ValidationError):
+        SSLCSRImport(certificate_content=_CERT_PEM, is_global=True, usage_type="both")

backend/tests/test_csr_router_auth.py

@@ -0,0 +1,66 @@
+"""
+v1.9.0 CSR creation — behavioral auth tests for /api/ssl/csrs endpoints
+(pattern: test_ssl_list_endpoint_auth.py).
+
+Every CSR endpoint must refuse unauthenticated / garbage-token requests.
+The CSR detail route additionally must never 200 without auth because it
+returns the CSR PEM; no endpoint ever returns the private key, but auth is
+the first line regardless.
+"""
+import pytest
+
+_VALID_CREATE_BODY = {
+    "name": "auth-test-csr",
+    "common_name": "www.example.com",
+}
+
+_VALID_IMPORT_BODY = {
+    "certificate_content": (
+        "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----"
+    ),
+    "is_global": True,
+}
+
+_ENDPOINTS = [
+    ("get", "/api/ssl/csrs", None),
+    ("get", "/api/ssl/csrs/1", None),
+    ("post", "/api/ssl/csrs", _VALID_CREATE_BODY),
+    ("post", "/api/ssl/csrs/1/import", _VALID_IMPORT_BODY),
+    ("delete", "/api/ssl/csrs/1", None),
+]
+
+
+@pytest.mark.parametrize("method,path,body", _ENDPOINTS)
+def test_csr_endpoint_unauthenticated_rejected(client, method, path, body):
+    """No Authorization header → endpoint must refuse the request."""
+    res = getattr(client, method)(path, json=body) if body is not None else getattr(client, method)(path)
+    assert res.status_code in (401, 403, 422), (
+        f"{method.upper()} {path} without Authorization returned "
+        f"{res.status_code} — anonymous access to CSR data must not be "
+        f"possible. Body: {res.text[:200]}"
+    )
+    if res.status_code == 200:  # defensive, mirrors the R18 test style
+        data = res.json()
+        assert not data, "CSR endpoint returned data without auth"
+
+
+@pytest.mark.parametrize("method,path,body", _ENDPOINTS)
+def test_csr_endpoint_invalid_token_rejected(client, method, path, body):
+    """Garbage token → endpoint must refuse the request."""
+    headers = {"Authorization": "Bearer not-a-valid-jwt"}
+    if body is not None:
+        res = getattr(client, method)(path, json=body, headers=headers)
+    else:
+        res = getattr(client, method)(path, headers=headers)
+    assert res.status_code in (401, 403, 422), (
+        f"{method.upper()} {path} with an invalid token returned {res.status_code}"
+    )
+
+
+def test_csr_routes_are_registered(client):
+    """The router must actually be mounted — a 404 would make the auth tests
+    above pass vacuously."""
+    res = client.get("/api/ssl/csrs")
+    assert res.status_code != 404, (
+        "GET /api/ssl/csrs returned 404 — csr_router is not registered in main.py"
+    )

backend/tests/test_csr_service_crypto.py

@@ -0,0 +1,171 @@
+"""
+v1.9.0 CSR creation — pure-crypto tests for services/csr_service.py.
+
+No mocks: every algorithm's output must parse with `cryptography` and the
+CSR's public key must match the generated private key (the property the
+whole import flow depends on).
+"""
+from types import SimpleNamespace
+
+import pytest
+
+from cryptography import x509
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import ec, rsa
+from cryptography.x509.oid import ExtensionOID, NameOID
+
+from services.csr_service import csr_row_to_dict, diff_domains, generate_csr_bundle
+
+
+def _payload(**overrides):
+    base = dict(
+        name="test-csr",
+        common_name="www.example.com",
+        organization=None,
+        organizational_unit=None,
+        locality=None,
+        state=None,
+        country=None,
+        email=None,
+        sans=[],
+        key_algorithm="rsa-2048",
+    )
+    base.update(overrides)
+    return SimpleNamespace(**base)
+
+
+def _spki(key):
+    return key.public_key().public_bytes(
+        serialization.Encoding.DER,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+
+
+@pytest.mark.parametrize(
+    "algo,key_cls,key_check",
+    [
+        ("rsa-2048", rsa.RSAPrivateKey, lambda k: k.key_size == 2048),
+        ("rsa-4096", rsa.RSAPrivateKey, lambda k: k.key_size == 4096),
+        ("ecdsa-p256", ec.EllipticCurvePrivateKey, lambda k: k.curve.name == "secp256r1"),
+        ("ecdsa-p384", ec.EllipticCurvePrivateKey, lambda k: k.curve.name == "secp384r1"),
+    ],
+)
+def test_generate_bundle_all_algorithms(algo, key_cls, key_check):
+    bundle = generate_csr_bundle(_payload(key_algorithm=algo))
+
+    csr = x509.load_pem_x509_csr(bundle["csr_pem"].encode())
+    key = serialization.load_pem_private_key(
+        bundle["private_key_pem"].encode(), password=None
+    )
+
+    assert isinstance(key, key_cls)
+    assert key_check(key)
+    # The CSR must be signed by exactly this key.
+    csr_spki = csr.public_key().public_bytes(
+        serialization.Encoding.DER,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    assert csr_spki == _spki(key)
+    assert csr.is_signature_valid
+    # PKCS8, unencrypted — the agent concatenates cert+key into one PEM and
+    # HAProxy cannot read passphrase-protected keys.
+    assert bundle["private_key_pem"].startswith("-----BEGIN PRIVATE KEY-----")
+
+
+def test_subject_contains_all_provided_fields():
+    bundle = generate_csr_bundle(_payload(
+        organization="Example Corp",
+        organizational_unit="IT",
+        locality="Istanbul",
+        state="Marmara",
+        country="TR",
+        email="ops@example.com",
+    ))
+    csr = x509.load_pem_x509_csr(bundle["csr_pem"].encode())
+
+    def _one(oid):
+        attrs = csr.subject.get_attributes_for_oid(oid)
+        return attrs[0].value if attrs else None
+
+    assert _one(NameOID.COMMON_NAME) == "www.example.com"
+    assert _one(NameOID.ORGANIZATION_NAME) == "Example Corp"
+    assert _one(NameOID.ORGANIZATIONAL_UNIT_NAME) == "IT"
+    assert _one(NameOID.LOCALITY_NAME) == "Istanbul"
+    assert _one(NameOID.STATE_OR_PROVINCE_NAME) == "Marmara"
+    assert _one(NameOID.COUNTRY_NAME) == "TR"
+    assert _one(NameOID.EMAIL_ADDRESS) == "ops@example.com"
+    assert bundle["subject"] == {
+        "O": "Example Corp", "OU": "IT", "L": "Istanbul",
+        "ST": "Marmara", "C": "TR", "emailAddress": "ops@example.com",
+    }
+
+
+def test_subject_omits_empty_fields():
+    bundle = generate_csr_bundle(_payload())
+    csr = x509.load_pem_x509_csr(bundle["csr_pem"].encode())
+    assert not csr.subject.get_attributes_for_oid(NameOID.ORGANIZATION_NAME)
+    assert bundle["subject"] == {}
+
+
+def test_sans_cn_first_and_deduped():
+    bundle = generate_csr_bundle(_payload(
+        common_name="www.example.com",
+        sans=["api.example.com", "www.example.com", "api.example.com", "cdn.example.com"],
+    ))
+    assert bundle["sans"] == ["www.example.com", "api.example.com", "cdn.example.com"]
+
+    csr = x509.load_pem_x509_csr(bundle["csr_pem"].encode())
+    san_ext = csr.extensions.get_extension_for_oid(
+        ExtensionOID.SUBJECT_ALTERNATIVE_NAME
+    )
+    dns_names = san_ext.value.get_values_for_type(x509.DNSName)
+    assert dns_names == ["www.example.com", "api.example.com", "cdn.example.com"]
+
+
+def test_wildcard_common_name_flows_into_san():
+    bundle = generate_csr_bundle(_payload(common_name="*.example.com"))
+    csr = x509.load_pem_x509_csr(bundle["csr_pem"].encode())
+    san_ext = csr.extensions.get_extension_for_oid(
+        ExtensionOID.SUBJECT_ALTERNATIVE_NAME
+    )
+    assert san_ext.value.get_values_for_type(x509.DNSName) == ["*.example.com"]
+
+
+def test_diff_domains_reports_added_and_dropped():
+    warnings = diff_domains(
+        ["www.example.com", "api.example.com"],
+        ["WWW.example.com", "cdn.example.com"],
+    )
+    assert len(warnings) == 2
+    added = next(w for w in warnings if "added" in w)
+    dropped = next(w for w in warnings if "dropped" in w)
+    assert "cdn.example.com" in added
+    assert "api.example.com" in dropped
+    # Case-insensitive: www must NOT be reported in either direction.
+    assert "www.example.com" not in added
+    assert "www.example.com" not in dropped
+
+
+def test_diff_domains_identical_sets_yield_no_warnings():
+    assert diff_domains(["a.example.com"], ["A.EXAMPLE.COM"]) == []
+    assert diff_domains([], []) == []
+
+
+def test_csr_row_to_dict_never_exposes_private_key():
+    row = {
+        "id": 1,
+        "name": "x",
+        "private_key_pem": "-----BEGIN PRIVATE KEY-----\nSECRET\n-----END PRIVATE KEY-----",
+        "csr_pem": "-----BEGIN CERTIFICATE REQUEST-----\nX\n-----END CERTIFICATE REQUEST-----",
+        "subject": '{"O": "Example"}',
+        "sans": '["a.example.com"]',
+    }
+    out = csr_row_to_dict(row)
+    assert "private_key_pem" not in out
+    assert "csr_pem" not in out          # lists exclude the PEM
+    assert out["subject"] == {"O": "Example"}
+    assert out["sans"] == ["a.example.com"]
+
+    detail = csr_row_to_dict(row, include_pem=True)
+    assert "private_key_pem" not in detail  # NEVER, even on detail
+    assert detail["csr_pem"].startswith("-----BEGIN CERTIFICATE REQUEST-----")

backend/version.json

@@ -1,5 +1,5 @@
 {
-  "version": "1.8.10",
-  "releaseName": "Security hardening — RCE, missing-auth and SSRF advisories (GHSA-7rhv/3p5c/3vh4)",
-  "releaseDate": "2026-07-20"
+  "version": "1.9.0",
+  "releaseName": "CSR creation — in-app key+CSR generation and signed-certificate import",
+  "releaseDate": "2026-08-04"
 }

frontend/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "haproxy-openmanager-frontend",
-  "version": "1.8.10",
+  "version": "1.9.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "haproxy-openmanager-frontend",
-      "version": "1.8.10",
+      "version": "1.9.0",
       "license": "AGPL-3.0-or-later",
       "dependencies": {
         "@ant-design/icons": "^5.0.0",

frontend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "haproxy-openmanager-frontend",
-  "version": "1.8.10",
+  "version": "1.9.0",
   "description": "HAProxy Load Balancer Management UI",
   "license": "AGPL-3.0-or-later",
   "dependencies": {

frontend/src/components/CSRManagement.js

@@ -0,0 +1,886 @@
+import React, { useState, useEffect, useCallback } from 'react';
+import {
+  Card, Table, Button, Modal, Form, Input, Space, message,
+  Popconfirm, Tag, Tooltip, Row, Col, Typography, Alert, Select,
+  Collapse, Descriptions, Badge
+} from 'antd';
+import {
+  PlusOutlined, ReloadOutlined, DeleteOutlined, EyeOutlined,
+  DownloadOutlined, CopyOutlined, FileProtectOutlined, ImportOutlined,
+  KeyOutlined
+} from '@ant-design/icons';
+import axios from 'axios';
+import { useCluster } from '../contexts/ClusterContext';
+import { extractApiError } from '../utils/apiError';
+import { antdDomainRule, antdDomainsListRule } from '../utils/validation';
+
+const { Text } = Typography;
+const { TextArea } = Input;
+
+const KEY_ALGORITHM_OPTIONS = [
+  { value: 'rsa-2048', label: 'RSA 2048 (recommended)' },
+  { value: 'rsa-4096', label: 'RSA 4096' },
+  { value: 'ecdsa-p256', label: 'ECDSA P-256' },
+  { value: 'ecdsa-p384', label: 'ECDSA P-384' },
+];
+
+const KEY_ALGORITHM_LABELS = {
+  'rsa-2048': 'RSA 2048',
+  'rsa-4096': 'RSA 4096',
+  'ecdsa-p256': 'ECDSA P-256',
+  'ecdsa-p384': 'ECDSA P-384',
+};
+
+// CSR creation (v1.9.0): generate the private key + CSR server-side, submit
+// the CSR to an external CA, then import the signed certificate. The private
+// key never leaves the backend — this component only ever handles the CSR
+// PEM and the CA's certificate response.
+const CSRManagement = ({ onCertificateImported }) => {
+  const { clusters } = useCluster();
+  const [csrs, setCsrs] = useState([]);
+  const [loading, setLoading] = useState(false);
+  const [createModalOpen, setCreateModalOpen] = useState(false);
+  const [creating, setCreating] = useState(false);
+  const [viewCsr, setViewCsr] = useState(null);
+  const [importCsr, setImportCsr] = useState(null);
+  const [importing, setImporting] = useState(false);
+  const [createForm] = Form.useForm();
+  const [importForm] = Form.useForm();
+
+  const fetchCsrs = useCallback(async () => {
+    setLoading(true);
+    try {
+      const response = await axios.get('/api/ssl/csrs', {
+        headers: {
+          'Cache-Control': 'no-cache, no-store, must-revalidate',
+          'Pragma': 'no-cache',
+        },
+      });
+      setCsrs(Array.isArray(response.data) ? response.data : []);
+    } catch (error) {
+      console.error('Error fetching CSRs:', error);
+      message.error(extractApiError(error, 'Failed to fetch CSRs'));
+    } finally {
+      setLoading(false);
+    }
+  }, []);
+
+  useEffect(() => {
+    fetchCsrs();
+  }, [fetchCsrs]);
+
+  const handleCreate = async (values) => {
+    setCreating(true);
+    try {
+      const payload = {
+        name: values.name,
+        common_name: values.common_name,
+        sans: values.sans || [],
+        key_algorithm: values.key_algorithm || 'rsa-2048',
+        organization: values.organization || null,
+        organizational_unit: values.organizational_unit || null,
+        locality: values.locality || null,
+        state: values.state || null,
+        country: values.country || null,
+        email: values.email || null,
+      };
+      const response = await axios.post('/api/ssl/csrs', payload);
+      message.success(
+        <div>
+          <strong>CSR '{values.name}' created</strong>
+          <br />
+          <small>Submit the CSR to your Certificate Authority for signing.</small>
+        </div>,
+        5
+      );
+      setCreateModalOpen(false);
+      createForm.resetFields();
+      fetchCsrs();
+      // Open the view modal immediately so the operator can copy/download
+      // the CSR PEM in one round trip.
+      if (response.data?.csr) {
+        setViewCsr(response.data.csr);
+      }
+    } catch (error) {
+      console.error('Error creating CSR:', error);
+      message.error(extractApiError(error, 'Failed to create CSR'));
+    } finally {
+      setCreating(false);
+    }
+  };
+
+  const handleView = async (record) => {
+    try {
+      const response = await axios.get(`/api/ssl/csrs/${record.id}`);
+      setViewCsr(response.data);
+    } catch (error) {
+      console.error('Error fetching CSR details:', error);
+      message.error(extractApiError(error, 'Failed to fetch CSR details'));
+    }
+  };
+
+  const handleDuplicate = (record) => {
+    const subject = record.subject || {};
+    createForm.setFieldsValue({
+      name: `${record.name}-new`,
+      common_name: record.common_name,
+      sans: (record.sans || []).filter((s) => s !== record.common_name),
+      key_algorithm: record.key_algorithm || 'rsa-2048',
+      organization: subject.O || undefined,
+      organizational_unit: subject.OU || undefined,
+      locality: subject.L || undefined,
+      state: subject.ST || undefined,
+      country: subject.C || undefined,
+      email: subject.emailAddress || undefined,
+    });
+    setCreateModalOpen(true);
+  };
+
+  const handleDelete = async (record) => {
+    try {
+      const response = await axios.delete(`/api/ssl/csrs/${record.id}`);
+      message.success(response.data?.message || `CSR '${record.name}' deleted`);
+      fetchCsrs();
+    } catch (error) {
+      console.error('Error deleting CSR:', error);
+      message.error(extractApiError(error, 'Failed to delete CSR'));
+    }
+  };
+
+  const handleImport = async (values) => {
+    if (!importCsr) return;
+    setImporting(true);
+    try {
+      const isGlobal = values.ssl_type === 'global';
+      const payload = {
+        certificate_content: values.certificate_content,
+        chain_content: values.chain_content || null,
+        usage_type: values.usage_type || 'frontend',
+        is_global: isGlobal,
+        cluster_ids: isGlobal ? null : values.cluster_ids,
+        name: values.name_override ? values.name_override.trim() : null,
+      };
+      const response = await axios.post(
+        `/api/ssl/csrs/${importCsr.id}/import`,
+        payload
+      );
+      const warnings = response.data?.warnings || [];
+      if (warnings.length > 0) {
+        Modal.warning({
+          title: 'Certificate imported with warnings',
+          width: 560,
+          content: (
+            <ul style={{ paddingLeft: 18, marginTop: 8 }}>
+              {warnings.map((w, i) => (
+                <li key={i}>{w}</li>
+              ))}
+            </ul>
+          ),
+        });
+      }
+      message.success(
+        <div>
+          <strong>Certificate imported successfully</strong>
+          <br />
+          <small>Go to Apply Management to deploy it to the cluster(s).</small>
+        </div>,
+        6
+      );
+      setImportCsr(null);
+      importForm.resetFields();
+      fetchCsrs();
+      if (onCertificateImported) {
+        onCertificateImported();
+      }
+    } catch (error) {
+      console.error('Error importing signed certificate:', error);
+      message.error(extractApiError(error, 'Failed to import certificate'));
+    } finally {
+      setImporting(false);
+    }
+  };
+
+  const downloadCsrPem = (csr) => {
+    if (!csr?.csr_pem) return;
+    const blob = new Blob([csr.csr_pem], { type: 'application/pkcs10;charset=utf-8' });
+    const url = URL.createObjectURL(blob);
+    const link = document.createElement('a');
+    link.href = url;
+    link.download = `${csr.name}.csr`;
+    document.body.appendChild(link);
+    link.click();
+    document.body.removeChild(link);
+    URL.revokeObjectURL(url);
+  };
+
+  const copyCsrPem = (csr) => {
+    if (!csr?.csr_pem) return;
+    if (navigator.clipboard && navigator.clipboard.writeText) {
+      navigator.clipboard
+        .writeText(csr.csr_pem)
+        .then(() => message.success('CSR PEM copied to clipboard'))
+        .catch(() => message.error('Failed to copy CSR PEM'));
+    } else {
+      message.warning('Clipboard is not available in this browser');
+    }
+  };
+
+  const columns = [
+    {
+      title: 'CSR',
+      dataIndex: 'name',
+      key: 'name',
+      render: (text, record) => (
+        <Space>
+          <FileProtectOutlined style={{ color: '#1677ff' }} />
+          <div>
+            <strong>{text}</strong>
+            <br />
+            <Text type="secondary" style={{ fontSize: 12 }}>
+              {record.common_name}
+            </Text>
+          </div>
+        </Space>
+      ),
+    },
+    {
+      title: 'SANs',
+      dataIndex: 'sans',
+      key: 'sans',
+      render: (sans) => {
+        const list = Array.isArray(sans) ? sans : [];
+        if (list.length === 0) return <Text type="secondary">-</Text>;
+        const visible = list.slice(0, 2);
+        const rest = list.slice(2);
+        return (
+          <Space size={4} wrap>
+            {visible.map((d) => (
+              <Tag key={d}>{d}</Tag>
+            ))}
+            {rest.length > 0 && (
+              <Tooltip title={rest.join(', ')}>
+                <Tag>+{rest.length}</Tag>
+              </Tooltip>
+            )}
+          </Space>
+        );
+      },
+    },
+    {
+      title: 'Key',
+      dataIndex: 'key_algorithm',
+      key: 'key_algorithm',
+      render: (algo) => (
+        <Tag icon={<KeyOutlined />} color="geekblue">
+          {KEY_ALGORITHM_LABELS[algo] || algo}
+        </Tag>
+      ),
+    },
+    {
+      title: 'Status',
+      dataIndex: 'status',
+      key: 'status',
+      render: (status, record) => {
+        if (status === 'completed') {
+          return (
+            <div>
+              <Badge status="success" text="Imported" />
+              {record.certificate_name && (
+                <>
+                  <br />
+                  <Text type="secondary" style={{ fontSize: 12 }}>
+                    → {record.certificate_name}
+                  </Text>
+                </>
+              )}
+            </div>
+          );
+        }
+        return <Badge status="processing" text="Awaiting certificate" />;
+      },
+    },
+    {
+      title: 'Created',
+      dataIndex: 'created_at',
+      key: 'created_at',
+      render: (date, record) => (
+        <div>
+          {date
+            ? new Date(date).toLocaleString(undefined, {
+                year: 'numeric',
+                month: 'short',
+                day: 'numeric',
+                hour: '2-digit',
+                minute: '2-digit',
+              })
+            : '-'}
+          {record.created_by_username && (
+            <>
+              <br />
+              <Text type="secondary" style={{ fontSize: 12 }}>
+                by {record.created_by_username}
+              </Text>
+            </>
+          )}
+        </div>
+      ),
+    },
+    {
+      title: 'Actions',
+      key: 'actions',
+      render: (_, record) => (
+        <Space size="small">
+          <Tooltip title="View / download CSR">
+            <Button
+              type="text"
+              size="small"
+              icon={<EyeOutlined />}
+              onClick={() => handleView(record)}
+            />
+          </Tooltip>
+          {record.status === 'pending' && (
+            <Tooltip title="Import signed certificate">
+              <Button
+                type="primary"
+                size="small"
+                icon={<ImportOutlined />}
+                onClick={() => {
+                  importForm.resetFields();
+                  setImportCsr(record);
+                }}
+              >
+                Import
+              </Button>
+            </Tooltip>
+          )}
+          <Tooltip title="Duplicate (pre-fill a new CSR)">
+            <Button
+              type="text"
+              size="small"
+              icon={<CopyOutlined />}
+              onClick={() => handleDuplicate(record)}
+            />
+          </Tooltip>
+          <Popconfirm
+            title="Delete this CSR?"
+            description={
+              record.status === 'pending'
+                ? 'The private key will be permanently destroyed — any certificate later signed from this CSR becomes unusable.'
+                : 'Only the CSR history entry is removed — the imported certificate is not affected.'
+            }
+            onConfirm={() => handleDelete(record)}
+            okText="Delete"
+            okType="danger"
+            cancelText="Cancel"
+          >
+            <Tooltip title="Delete CSR">
+              <Button type="text" size="small" danger icon={<DeleteOutlined />} />
+            </Tooltip>
+          </Popconfirm>
+        </Space>
+      ),
+    },
+  ];
+
+  const pendingCount = csrs.filter((c) => c.status === 'pending').length;
+
+  return (
+    <div>
+      <Alert
+        type="info"
+        showIcon
+        style={{ marginBottom: 16 }}
+        message="Certificate Signing Requests for external CAs"
+        description="Generate a private key and CSR here, submit the CSR to your Certificate Authority, then import the signed certificate. The private key never leaves the server; the imported certificate goes through the normal Apply Management deployment flow."
+      />
+      <Row gutter={16} style={{ marginBottom: 16 }}>
+        <Col flex="auto">
+          {pendingCount > 0 && (
+            <Text type="secondary">
+              {pendingCount} CSR{pendingCount > 1 ? 's' : ''} awaiting a signed
+              certificate
+            </Text>
+          )}
+        </Col>
+        <Col>
+          <Space>
+            <Button icon={<ReloadOutlined />} onClick={fetchCsrs} loading={loading}>
+              Refresh
+            </Button>
+            <Button
+              type="primary"
+              icon={<PlusOutlined />}
+              onClick={() => {
+                createForm.resetFields();
+                setCreateModalOpen(true);
+              }}
+            >
+              Create CSR
+            </Button>
+          </Space>
+        </Col>
+      </Row>
+
+      <Card>
+        <Table
+          columns={columns}
+          dataSource={csrs}
+          rowKey="id"
+          loading={loading}
+          pagination={{
+            showSizeChanger: true,
+            showQuickJumper: true,
+            showTotal: (total) => `Total ${total} CSRs`,
+          }}
+        />
+      </Card>
+
+      {/* Create CSR Modal */}
+      <Modal
+        title="Create Certificate Signing Request"
+        open={createModalOpen}
+        onCancel={() => {
+          setCreateModalOpen(false);
+          createForm.resetFields();
+        }}
+        footer={null}
+        width={700}
+        forceRender
+      >
+        <Form form={createForm} layout="vertical" onFinish={handleCreate}>
+          <Form.Item
+            name="name"
+            label="Name"
+            rules={[
+              { required: true, message: 'Please enter a CSR name' },
+              {
+                pattern: /^[a-zA-Z0-9_.-]+$/,
+                message:
+                  'Only letters, digits, underscore, hyphen and dot are allowed',
+              },
+              { max: 100, message: 'Name must be 100 characters or fewer' },
+              {
+                validator: (_, value) => {
+                  if (!value) return Promise.resolve();
+                  if (value.includes('..')) {
+                    return Promise.reject(new Error('Name must not contain ".."'));
+                  }
+                  if (value.startsWith('.') || value.startsWith('-')) {
+                    return Promise.reject(
+                      new Error('Name must not start with "." or "-"')
+                    );
+                  }
+                  return Promise.resolve();
+                },
+              },
+            ]}
+            extra="Becomes the certificate name and file path at import: /etc/ssl/haproxy/{name}.pem"
+          >
+            <Input placeholder="e.g. www-example-com" />
+          </Form.Item>
+
+          <Form.Item
+            name="common_name"
+            label="Common Name (CN)"
+            rules={[
+              { required: true, message: 'Please enter the Common Name' },
+              antdDomainRule,
+              { max: 64, message: 'Common Name must be 64 characters or fewer' },
+            ]}
+            extra="The primary domain, e.g. www.example.com or *.example.com"
+          >
+            <Input placeholder="www.example.com" />
+          </Form.Item>
+
+          <Form.Item
+            name="sans"
+            label="Subject Alternative Names (SANs)"
+            rules={[antdDomainsListRule]}
+            extra="Additional DNS names — the Common Name is included automatically"
+          >
+            <Select
+              mode="tags"
+              tokenSeparators={[',', ' ']}
+              placeholder="api.example.com, cdn.example.com"
+              open={false}
+              suffixIcon={null}
+            />
+          </Form.Item>
+
+          <Form.Item
+            name="key_algorithm"
+            label="Key Algorithm"
+            initialValue="rsa-2048"
+            rules={[{ required: true }]}
+          >
+            <Select options={KEY_ALGORITHM_OPTIONS} />
+          </Form.Item>
+
+          <Collapse
+            style={{ marginBottom: 16 }}
+            items={[
+              {
+                key: 'subject',
+                label: 'Subject details (optional)',
+                children: (
+                  <>
+                    <Row gutter={12}>
+                      <Col span={12}>
+                        <Form.Item
+                          name="organization"
+                          label="Organization (O)"
+                          rules={[{ max: 64 }]}
+                        >
+                          <Input placeholder="Example Corp" />
+                        </Form.Item>
+                      </Col>
+                      <Col span={12}>
+                        <Form.Item
+                          name="organizational_unit"
+                          label="Organizational Unit (OU)"
+                          rules={[{ max: 64 }]}
+                        >
+                          <Input placeholder="IT Department" />
+                        </Form.Item>
+                      </Col>
+                    </Row>
+                    <Row gutter={12}>
+                      <Col span={8}>
+                        <Form.Item name="locality" label="Locality (L)" rules={[{ max: 64 }]}>
+                          <Input placeholder="Istanbul" />
+                        </Form.Item>
+                      </Col>
+                      <Col span={8}>
+                        <Form.Item name="state" label="State / Province (ST)" rules={[{ max: 64 }]}>
+                          <Input placeholder="Marmara" />
+                        </Form.Item>
+                      </Col>
+                      <Col span={8}>
+                        <Form.Item
+                          name="country"
+                          label="Country (C)"
+                          rules={[
+                            {
+                              pattern: /^[A-Za-z]{2}$/,
+                              message: 'Exactly 2 letters (e.g. TR, US)',
+                            },
+                          ]}
+                        >
+                          <Input placeholder="TR" maxLength={2} />
+                        </Form.Item>
+                      </Col>
+                    </Row>
+                    <Form.Item
+                      name="email"
+                      label="Email"
+                      rules={[{ type: 'email', message: 'Invalid email address' }]}
+                    >
+                      <Input placeholder="ops@example.com" />
+                    </Form.Item>
+                  </>
+                ),
+              },
+            ]}
+          />
+
+          <Alert
+            type="info"
+            showIcon
+            style={{ marginBottom: 16 }}
+            message="🔐 The private key is generated and stored server-side"
+            description="You will only receive the CSR to hand to your CA. After the signed certificate is imported, the key is available on the certificate itself."
+          />
+
+          <Form.Item style={{ textAlign: 'right', marginBottom: 0 }}>
+            <Space>
+              <Button
+                onClick={() => {
+                  setCreateModalOpen(false);
+                  createForm.resetFields();
+                }}
+              >
+                Cancel
+              </Button>
+              <Button type="primary" htmlType="submit" loading={creating}>
+                Generate CSR
+              </Button>
+            </Space>
+          </Form.Item>
+        </Form>
+      </Modal>
+
+      {/* View CSR Modal */}
+      <Modal
+        title={
+          <Space>
+            <FileProtectOutlined />
+            {viewCsr ? `CSR: ${viewCsr.name}` : 'CSR'}
+          </Space>
+        }
+        open={!!viewCsr}
+        onCancel={() => setViewCsr(null)}
+        width={760}
+        footer={[
+          <Button key="close" onClick={() => setViewCsr(null)}>
+            Close
+          </Button>,
+        ]}
+      >
+        {viewCsr && (
+          <div>
+            <Descriptions size="small" column={2} bordered style={{ marginBottom: 12 }}>
+              <Descriptions.Item label="Common Name" span={2}>
+                {viewCsr.common_name}
+              </Descriptions.Item>
+              <Descriptions.Item label="Key">
+                {KEY_ALGORITHM_LABELS[viewCsr.key_algorithm] || viewCsr.key_algorithm}
+              </Descriptions.Item>
+              <Descriptions.Item label="Status">
+                {viewCsr.status === 'completed' ? (
+                  <Badge status="success" text="Imported" />
+                ) : (
+                  <Badge status="processing" text="Awaiting certificate" />
+                )}
+              </Descriptions.Item>
+              {viewCsr.subject && Object.keys(viewCsr.subject).length > 0 && (
+                <Descriptions.Item label="Subject" span={2}>
+                  {Object.entries(viewCsr.subject)
+                    .map(([k, v]) => `${k}=${v}`)
+                    .join(', ')}
+                </Descriptions.Item>
+              )}
+            </Descriptions>
+
+            {Array.isArray(viewCsr.sans) && viewCsr.sans.length > 0 && (
+              <div style={{ marginBottom: 12 }}>
+                <Text strong>SANs: </Text>
+                <Space size={4} wrap>
+                  {viewCsr.sans.map((d) => (
+                    <Tag key={d}>{d}</Tag>
+                  ))}
+                </Space>
+              </div>
+            )}
+
+            <Row justify="space-between" align="middle" style={{ marginBottom: 8 }}>
+              <Col>
+                <Text strong>CSR (PEM)</Text>
+              </Col>
+              <Col>
+                <Space>
+                  <Button
+                    size="small"
+                    icon={<CopyOutlined />}
+                    onClick={() => copyCsrPem(viewCsr)}
+                  >
+                    Copy
+                  </Button>
+                  <Button
+                    size="small"
+                    type="primary"
+                    icon={<DownloadOutlined />}
+                    onClick={() => downloadCsrPem(viewCsr)}
+                  >
+                    Download .csr
+                  </Button>
+                </Space>
+              </Col>
+            </Row>
+            <TextArea
+              value={viewCsr.csr_pem}
+              rows={12}
+              readOnly
+              style={{ fontFamily: 'monospace', fontSize: 12 }}
+            />
+            {viewCsr.status !== 'completed' && (
+              <Alert
+                type="info"
+                showIcon
+                style={{ marginTop: 12 }}
+                message="Next step"
+                description="Submit this CSR to your Certificate Authority. When you receive the signed certificate, come back and click Import on this CSR."
+              />
+            )}
+          </div>
+        )}
+      </Modal>
+
+      {/* Import Signed Certificate Modal */}
+      <Modal
+        title={
+          <Space>
+            <ImportOutlined />
+            {importCsr ? `Import Signed Certificate — ${importCsr.name}` : 'Import'}
+          </Space>
+        }
+        open={!!importCsr}
+        onCancel={() => {
+          setImportCsr(null);
+          importForm.resetFields();
+        }}
+        footer={null}
+        width={800}
+      >
+        {importCsr && (
+          <div>
+            <Alert
+              type="info"
+              showIcon
+              style={{ marginBottom: 16 }}
+              message={`CSR: ${importCsr.name} (CN: ${importCsr.common_name})`}
+              description="Paste the certificate your CA issued for this CSR. It will be verified against the stored private key before anything is saved."
+            />
+            <Form
+              form={importForm}
+              layout="vertical"
+              onFinish={handleImport}
+              initialValues={{ ssl_type: 'cluster', usage_type: 'frontend' }}
+            >
+              <Row gutter={12}>
+                <Col span={12}>
+                  <Form.Item
+                    name="usage_type"
+                    label="Usage Type"
+                    rules={[{ required: true }]}
+                  >
+                    <Select
+                      options={[
+                        { value: 'frontend', label: 'Frontend SSL (HTTPS termination)' },
+                        { value: 'server', label: 'Server SSL (backend verification)' },
+                      ]}
+                    />
+                  </Form.Item>
+                </Col>
+                <Col span={12}>
+                  <Form.Item name="ssl_type" label="Scope" rules={[{ required: true }]}>
+                    <Select
+                      options={[
+                        { value: 'global', label: 'Global (all clusters)' },
+                        { value: 'cluster', label: 'Cluster-specific' },
+                      ]}
+                    />
+                  </Form.Item>
+                </Col>
+              </Row>
+
+              <Form.Item
+                noStyle
+                shouldUpdate={(prev, cur) => prev.ssl_type !== cur.ssl_type}
+              >
+                {({ getFieldValue }) =>
+                  getFieldValue('ssl_type') === 'cluster' && (
+                    <Form.Item
+                      name="cluster_ids"
+                      label="Clusters"
+                      rules={[
+                        { required: true, message: 'Select at least one cluster' },
+                      ]}
+                    >
+                      <Select
+                        mode="multiple"
+                        placeholder="Select cluster(s)"
+                        options={(clusters || []).map((c) => ({
+                          value: c.id,
+                          label: c.name,
+                        }))}
+                      />
+                    </Form.Item>
+                  )
+                }
+              </Form.Item>
+
+              <Form.Item
+                name="certificate_content"
+                label="Signed Certificate (PEM)"
+                rules={[
+                  { required: true, message: 'Please paste the signed certificate' },
+                  {
+                    validator: (_, value) => {
+                      if (!value) return Promise.resolve();
+                      if (
+                        value.includes('-----BEGIN CERTIFICATE-----') &&
+                        value.includes('-----END CERTIFICATE-----')
+                      ) {
+                        return Promise.resolve();
+                      }
+                      return Promise.reject(
+                        new Error('Certificate must be in PEM format')
+                      );
+                    },
+                  },
+                ]}
+              >
+                <TextArea
+                  rows={8}
+                  placeholder={'-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----'}
+                  style={{ fontFamily: 'monospace', fontSize: 12 }}
+                />
+              </Form.Item>
+
+              <Form.Item
+                name="name_override"
+                label="Certificate name override (optional)"
+                rules={[
+                  {
+                    pattern: /^[a-zA-Z0-9_.-]+$/,
+                    message:
+                      'Only letters, digits, underscore, hyphen and dot are allowed',
+                  },
+                  { max: 100, message: 'Name must be 100 characters or fewer' },
+                ]}
+                extra={`Leave empty to use the CSR name ('${importCsr.name}'). Use this only if that name is now taken by another certificate.`}
+              >
+                <Input placeholder={importCsr.name} />
+              </Form.Item>
+
+              <Form.Item
+                name="chain_content"
+                label="Certificate Chain (PEM, optional)"
+                rules={[
+                  {
+                    validator: (_, value) => {
+                      if (!value || !value.trim()) return Promise.resolve();
+                      if (
+                        value.includes('-----BEGIN CERTIFICATE-----') &&
+                        value.includes('-----END CERTIFICATE-----')
+                      ) {
+                        return Promise.resolve();
+                      }
+                      return Promise.reject(
+                        new Error('Certificate chain must be in PEM format')
+                      );
+                    },
+                  },
+                ]}
+              >
+                <TextArea
+                  rows={4}
+                  placeholder={'-----BEGIN CERTIFICATE-----\n(intermediate CA)\n-----END CERTIFICATE-----'}
+                  style={{ fontFamily: 'monospace', fontSize: 12 }}
+                />
+              </Form.Item>
+
+              <Form.Item style={{ textAlign: 'right', marginBottom: 0 }}>
+                <Space>
+                  <Button
+                    onClick={() => {
+                      setImportCsr(null);
+                      importForm.resetFields();
+                    }}
+                  >
+                    Cancel
+                  </Button>
+                  <Button type="primary" htmlType="submit" loading={importing}>
+                    Import Certificate
+                  </Button>
+                </Space>
+              </Form.Item>
+            </Form>
+          </div>
+        )}
+      </Modal>
+    </div>
+  );
+};
+
+export default CSRManagement;

frontend/src/components/SSLManagement.js

@@ -13,7 +13,7 @@ import {
   PlayCircleOutlined, EditOutlined,
   CloudServerOutlined, CheckCircleOutlined, SyncOutlined,
   ExclamationCircleOutlined, CloseCircleOutlined, ClockCircleOutlined,
-  ThunderboltOutlined
+  ThunderboltOutlined, FileProtectOutlined
 } from '@ant-design/icons';
 import axios from 'axios';
 import { useSearchParams } from 'react-router-dom';
@@ -22,6 +22,7 @@ import { useProgress } from '../contexts/ProgressContext';
 import { formatEntityForSync } from '../utils/agentSync';
 import { extractApiError } from '../utils/apiError';
 import ACMEAutomation from './ACMEAutomation';
+import CSRManagement from './CSRManagement';
 
 const { Title, Text } = Typography;
 const { TextArea } = Input;
@@ -643,12 +644,21 @@ const SSLManagement = () => {
       title: 'Source',
       dataIndex: 'source',
       key: 'source',
-      render: (source) => (
-        <Tag color={source === 'letsencrypt' ? 'green' : 'default'}
-             icon={source === 'letsencrypt' ? <SafetyCertificateOutlined /> : null}>
-          {source === 'letsencrypt' ? 'Auto (ACME)' : 'Manual'}
-        </Tag>
-      ),
+      render: (source) => {
+        if (source === 'csr') {
+          return (
+            <Tag color="blue" icon={<FileProtectOutlined />}>
+              CSR
+            </Tag>
+          );
+        }
+        return (
+          <Tag color={source === 'letsencrypt' ? 'green' : 'default'}
+               icon={source === 'letsencrypt' ? <SafetyCertificateOutlined /> : null}>
+            {source === 'letsencrypt' ? 'Auto (ACME)' : 'Manual'}
+          </Tag>
+        );
+      },
     },
     {
       title: 'Sync Status',
@@ -1020,6 +1030,11 @@ const SSLManagement = () => {
             label: <span><ThunderboltOutlined /> ACME Automation</span>,
             children: <ACMEAutomation />,
           },
+          {
+            key: 'csr',
+            label: <span><FileProtectOutlined /> CSR</span>,
+            children: <CSRManagement onCertificateImported={fetchCertificates} />,
+          },
         ]}
       />