231 lines
7.2 KiB
Python
231 lines
7.2 KiB
Python
"""PKI management: internal CA + server certificates (spec sections 16-19).
|
|
|
|
All endpoints require the PKI_ADMIN role (or super_admin). Private keys
|
|
never appear in any response — the PEM download endpoints expose only
|
|
the public certificate material.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from fastapi.responses import PlainTextResponse
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import audit
|
|
from app.crypto import encrypt_secret
|
|
from app.db import get_db
|
|
from app.deps import require_roles
|
|
from app.models import Certificate, ManagementRole
|
|
from app.pki import (
|
|
PkiError,
|
|
create_ca as pki_create_ca,
|
|
fingerprint as pki_fingerprint,
|
|
issue_server_certificate,
|
|
)
|
|
from app.schemas import CaCreate, CertificateOut, ServerCertCreate
|
|
|
|
router = APIRouter(prefix="/api/v1/pki", tags=["pki"])
|
|
|
|
PkiAdmin = Annotated[
|
|
object, Depends(require_roles(ManagementRole.PKI_ADMIN))
|
|
]
|
|
DbDep = Annotated[Session, Depends(get_db)]
|
|
|
|
|
|
def _get_certificate(db: Session, cert_id: uuid.UUID) -> Certificate:
|
|
cert = db.get(Certificate, cert_id)
|
|
if cert is None:
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "certificate not found")
|
|
return cert
|
|
|
|
|
|
@router.get("/ca", response_model=list[CertificateOut])
|
|
def list_cas(db: DbDep, _user: PkiAdmin):
|
|
"""All certificate authorities in the store."""
|
|
return db.scalars(
|
|
select(Certificate).where(Certificate.is_ca.is_(True))
|
|
).all()
|
|
|
|
|
|
@router.post("/ca", response_model=CertificateOut, status_code=201)
|
|
def create_ca(body: CaCreate, request: Request, db: DbDep, _user: PkiAdmin):
|
|
"""Create a self-signed internal CA (spec section 17)."""
|
|
try:
|
|
issued = pki_create_ca(
|
|
common_name=body.common_name,
|
|
validity_days=body.validity_days,
|
|
key_algorithm=body.key_algorithm,
|
|
organization=body.organization,
|
|
organizational_unit=body.organizational_unit,
|
|
country=body.country,
|
|
)
|
|
except PkiError as exc:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
|
|
|
|
ca = Certificate(
|
|
common_name=body.common_name,
|
|
certificate_pem=issued.certificate_pem,
|
|
private_key_pem=encrypt_secret(issued.private_key_pem),
|
|
expiry_date=issued.not_valid_after,
|
|
is_ca=True,
|
|
key_algorithm=body.key_algorithm,
|
|
serial_number=str(issued.serial_number),
|
|
dns_sans=[],
|
|
ip_sans=[],
|
|
fingerprint=pki_fingerprint(issued.certificate_pem),
|
|
status="issued",
|
|
)
|
|
db.add(ca)
|
|
db.flush()
|
|
audit.record(
|
|
db,
|
|
"CA_CREATED",
|
|
request=request,
|
|
object_type="certificate_authority",
|
|
object_id=str(ca.id),
|
|
new_value={"common_name": body.common_name},
|
|
)
|
|
return ca
|
|
|
|
|
|
@router.get("/certificates", response_model=list[CertificateOut])
|
|
def list_certificates(db: DbDep, _user: PkiAdmin):
|
|
"""Full certificate inventory (spec section 21), newest first."""
|
|
return db.scalars(
|
|
select(Certificate).order_by(Certificate.created_at.desc())
|
|
).all()
|
|
|
|
|
|
@router.post("/certificates", response_model=CertificateOut, status_code=201)
|
|
def issue_certificate(
|
|
body: ServerCertCreate, request: Request, db: DbDep, _user: PkiAdmin
|
|
):
|
|
"""Issue a server certificate signed by the given CA (spec section 19)."""
|
|
from app.crypto import decrypt_secret
|
|
|
|
ca = _get_certificate(db, body.ca_id)
|
|
if not ca.is_ca:
|
|
raise HTTPException(
|
|
status.HTTP_400_BAD_REQUEST, "the referenced certificate is not a CA"
|
|
)
|
|
if ca.private_key_pem is None:
|
|
raise HTTPException(
|
|
status.HTTP_409_CONFLICT,
|
|
"the CA private key is not held by the platform (offline CA); "
|
|
"it cannot issue certificates",
|
|
)
|
|
try:
|
|
ca_key_pem = decrypt_secret(ca.private_key_pem)
|
|
except ValueError as exc:
|
|
raise HTTPException(
|
|
status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
f"could not decrypt the CA key: {exc}",
|
|
) from exc
|
|
|
|
try:
|
|
issued = issue_server_certificate(
|
|
ca_certificate_pem=ca.certificate_pem,
|
|
ca_private_key_pem=ca_key_pem,
|
|
common_name=body.common_name,
|
|
validity_days=body.validity_days,
|
|
dns_sans=body.dns_sans,
|
|
ip_sans=body.ip_sans,
|
|
organization=body.organization,
|
|
organizational_unit=body.organizational_unit,
|
|
country=body.country,
|
|
)
|
|
except PkiError as exc:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
|
|
|
|
cert = Certificate(
|
|
common_name=body.common_name,
|
|
certificate_pem=issued.certificate_pem,
|
|
private_key_pem=encrypt_secret(issued.private_key_pem),
|
|
expiry_date=issued.not_valid_after,
|
|
is_ca=False,
|
|
parent_id=ca.id,
|
|
key_algorithm="ecdsa-p256",
|
|
serial_number=str(issued.serial_number),
|
|
dns_sans=body.dns_sans,
|
|
ip_sans=body.ip_sans,
|
|
fingerprint=pki_fingerprint(issued.certificate_pem),
|
|
status="issued",
|
|
)
|
|
db.add(cert)
|
|
db.flush()
|
|
audit.record(
|
|
db,
|
|
"CERTIFICATE_ISSUED",
|
|
request=request,
|
|
object_type="certificate",
|
|
object_id=str(cert.id),
|
|
new_value={
|
|
"common_name": body.common_name,
|
|
"ca": ca.common_name,
|
|
"dns_sans": body.dns_sans,
|
|
},
|
|
)
|
|
return cert
|
|
|
|
|
|
@router.get(
|
|
"/certificates/{cert_id}/download",
|
|
response_class=PlainTextResponse,
|
|
)
|
|
def download_certificate(cert_id: uuid.UUID, db: DbDep, _user: PkiAdmin):
|
|
"""Download the certificate (public part only) as PEM."""
|
|
cert = _get_certificate(db, cert_id)
|
|
return PlainTextResponse(
|
|
cert.certificate_pem,
|
|
media_type="application/x-pem-file",
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="{cert.common_name}.crt"'
|
|
},
|
|
)
|
|
|
|
|
|
@router.delete(
|
|
"/certificates/{cert_id}", status_code=status.HTTP_204_NO_CONTENT
|
|
)
|
|
def delete_certificate(
|
|
cert_id: uuid.UUID, request: Request, db: DbDep, _user: PkiAdmin
|
|
):
|
|
"""Remove a certificate from the store.
|
|
|
|
CAs with still-issued children are refused, and a certificate that a
|
|
listener is bound to cannot be deleted (the generated config would
|
|
reference a missing file).
|
|
"""
|
|
cert = _get_certificate(db, cert_id)
|
|
if cert.is_ca:
|
|
children = db.scalar(
|
|
select(Certificate.id).where(Certificate.parent_id == cert.id)
|
|
)
|
|
if children is not None:
|
|
raise HTTPException(
|
|
status.HTTP_409_CONFLICT,
|
|
"the CA still has issued certificates; delete them first",
|
|
)
|
|
from app.models import ListenerConfig
|
|
|
|
bound = db.scalar(
|
|
select(ListenerConfig.id).where(ListenerConfig.cert_id == cert.id)
|
|
)
|
|
if bound is not None:
|
|
raise HTTPException(
|
|
status.HTTP_409_CONFLICT,
|
|
"the certificate is bound to a listener; unbind it first",
|
|
)
|
|
db.delete(cert)
|
|
audit.record(
|
|
db,
|
|
"CERTIFICATE_DELETED",
|
|
request=request,
|
|
object_type="certificate",
|
|
object_id=str(cert_id),
|
|
)
|
|
return None |