mirror of
https://github.com/taylanbakircioglu/haproxy-openmanager.git
synced 2026-09-23 19:06:25 +00:00
mustafa.ulukaya
a6166d11b9
New /api/ssl/csrs endpoint group: generate a private key + CSR server-side
(RSA 2048/4096, ECDSA P-256/P-384; full subject + DNS SANs with wildcard
support), list/detail/delete CSRs, and import the CA-signed certificate.
- New ssl_csrs table (SCHEMA_VERSION 9 -> 10, additive + idempotent); the
migration re-raises on failure so a failed run is retried instead of being
stamped as applied.
- Import verifies the certificate against the stored key as a hard gate
(match=None is treated as an integrity error, not a lenient pass), rejects
malformed and expired certificates with 400, warns on SAN drift, and
creates a normal ssl_certificates row (source=csr, cluster_id=NULL,
last_config_status=PENDING) so it flows through the standard
Apply Management -> agent pull pipeline.
- Concurrency: FOR UPDATE row lock serialises double-import and
delete-during-import; a partial unique index reserves pending CSR names;
soft-deleted same-name certs are reactivated preserving the row id.
- Security: no CSR endpoint ever returns the private key (explicit column
lists, enforced by a static test); the key copy on the CSR row is NULLed
after import; ssl.create/read/delete permissions enforced on every
endpoint incl. reads; per-user rate limit on key generation, which runs
in a worker thread; csr_id and cluster_ids are int32-guarded.
- ssl_service: extract _prepare_cert_fields from create_cert_row (behaviour
unchanged, extraction tests untouched) and add stage_ssl_config_versions
reusing the exact ssl-{id}-create-{ts} version-name scheme.
- Tests: crypto round-trip for all four algorithms, model validation,
import-flow unit tests, endpoint auth/permission pinning, migration and
key-non-exposure static assertions.
67 lines
2.4 KiB
Python
67 lines
2.4 KiB
Python
"""
|
|
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"
|
|
)
|