Implement KCA: web reimplementation of XCA PKI workflows

- kaya backend with OIDC sessions, API tokens, Postgres/Tortoise models
- AES-256-GCM private key storage and full certificate/CSR/CRL domain logic
- Sycamore/WASM SPA for vaults, keys, requests, certificates, CRLs and templates
- imports/exports, XCA .xdb migration, Docker Compose stack and tests
This commit is contained in:
2026-10-23 02:06:55 +00:00
commit 4f3a632f4b
109 changed files with 21048 additions and 0 deletions
+7
View File
@@ -0,0 +1,7 @@
# kca-core
Pure, I/O-free PKI domain logic for KCA. It knows how to generate and parse
keys, build distinguished names, encode X.509 extensions, issue certificates,
create CSRs and CRLs, apply templates, and convert every supported import and
export format. It has no kaya, database or filesystem dependencies and is
unit-tested in isolation.
+18
View File
@@ -0,0 +1,18 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "kca-core"
version = "0.1.0"
description = "Pure PKI domain logic for KCA (keys, names, extensions, certificates, CSRs, CRLs, templates, imports/exports)"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"cryptography>=44",
"paramiko>=3.5",
]
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
@@ -0,0 +1 @@
"""Pure PKI domain logic for KCA."""
@@ -0,0 +1,522 @@
"""Certificate construction, parsing, validation and export helpers."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import dsa, ec, ed25519, rsa
from cryptography.x509.oid import ExtensionOID, NameOID
from .extensions import (
build_extensions,
eku_name,
extension_config_lines,
extension_rows,
extension_specs,
)
from .keys import PrivateKey, PublicKey, hash_algorithm, public_der
from .names import entries_from_name, name_from_entries, name_hash, one_line
#: XCA's "no well-defined expiration" sentinel.
FAR_FUTURE = datetime(9999, 12, 31, 23, 59, 59, tzinfo=timezone.utc)
MAX_SERIAL_BYTES = 20
@dataclass(frozen=True)
class CertificateInfo:
subject: x509.Name
issuer: x509.Name
serial_hex: str
not_before: datetime
not_after: datetime
is_ca: bool
path_length: Optional[int]
signature_algorithm: str
public_key_der: bytes
fingerprint_md5: str
fingerprint_sha1: str
fingerprint_sha256: str
subject_hash: str
issuer_hash: str
def _as_utc(value: datetime) -> datetime:
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
def normalize_datetime(value: datetime) -> datetime:
return _as_utc(value)
def serial_to_hex(value: int) -> str:
return format(value, "x")
def parse_serial_hex(value: str) -> int:
text = (value or "").strip().lower().replace(":", "")
if text.startswith("0x"):
text = text[2:]
if not text:
raise ValueError("serial number is empty")
if len(text) > MAX_SERIAL_BYTES * 2:
raise ValueError("serial number must be at most 20 octets")
serial = int(text, 16)
if serial <= 0:
raise ValueError("serial number must be positive")
return serial
def build_certificate(
*,
subject: x509.Name,
public_key: PublicKey,
issuer_cert: Optional[x509.Certificate],
issuer_private_key: PrivateKey,
serial: int,
not_before: datetime,
not_after: datetime,
extensions: Iterable[x509.Extension],
signature_hash: Optional[str] = "sha256",
) -> x509.Certificate:
"""Issue a certificate (self-signed when ``issuer_cert`` is None)."""
if serial <= 0:
raise ValueError("serial number must be positive")
if serial.bit_length() > MAX_SERIAL_BYTES * 8:
raise ValueError("serial number is too large")
not_before = normalize_datetime(not_before)
not_after = normalize_datetime(not_after)
if not_after <= not_before:
raise ValueError("notAfter must be later than notBefore")
issuer_name = issuer_cert.subject if issuer_cert is not None else subject
builder = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer_name)
.public_key(public_key)
.serial_number(serial)
.not_valid_before(not_before)
.not_valid_after(not_after)
)
for extension in extensions:
builder = builder.add_extension(extension.value, extension.critical)
algorithm = hash_algorithm(signature_hash, issuer_private_key)
return builder.sign(private_key=issuer_private_key, algorithm=algorithm)
def parse_certificate(data: bytes) -> x509.Certificate:
return x509.load_der_x509_certificate(data)
def certificate_der(cert: x509.Certificate) -> bytes:
return cert.public_bytes(serialization.Encoding.DER)
def certificate_pem(cert: x509.Certificate) -> str:
return cert.public_bytes(serialization.Encoding.PEM).decode("ascii")
def certificate_info(cert: x509.Certificate) -> CertificateInfo:
try:
bc = cert.extensions.get_extension_for_class(x509.BasicConstraints).value
is_ca = bool(bc.ca)
path_length = bc.path_length
except x509.ExtensionNotFound:
is_ca = False
path_length = None
return CertificateInfo(
subject=cert.subject,
issuer=cert.issuer,
serial_hex=serial_to_hex(cert.serial_number),
not_before=normalize_datetime(cert.not_valid_before_utc),
not_after=normalize_datetime(cert.not_valid_after_utc),
is_ca=is_ca,
path_length=path_length,
signature_algorithm=signature_algorithm_name(cert),
public_key_der=cert.public_key().public_bytes(
serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo
),
fingerprint_md5=fingerprint(cert, "md5"),
fingerprint_sha1=fingerprint(cert, "sha1"),
fingerprint_sha256=fingerprint(cert, "sha256"),
subject_hash=name_hash(cert.subject),
issuer_hash=name_hash(cert.issuer),
)
_SIGNATURE_OIDS = {
"1.2.840.113549.1.1.5": "sha1WithRSAEncryption",
"1.2.840.113549.1.1.11": "sha256WithRSAEncryption",
"1.2.840.113549.1.1.12": "sha384WithRSAEncryption",
"1.2.840.113549.1.1.13": "sha512WithRSAEncryption",
"1.2.840.113549.1.1.14": "sha224WithRSAEncryption",
"1.2.840.10045.4.1": "ecdsa-with-SHA1",
"1.2.840.10045.4.3.2": "ecdsa-with-SHA256",
"1.2.840.10045.4.3.3": "ecdsa-with-SHA384",
"1.2.840.10045.4.3.4": "ecdsa-with-SHA512",
"1.3.101.112": "Ed25519",
}
def signature_algorithm_name(cert: x509.Certificate) -> str:
algorithm = cert.signature_algorithm_oid
dotted = algorithm.dotted_string
if dotted in _SIGNATURE_OIDS:
return _SIGNATURE_OIDS[dotted]
name = getattr(algorithm, "name", None)
if name and name != "Undefined":
return name
return dotted
def fingerprint(cert: x509.Certificate, algorithm: str = "sha256") -> str:
data = certificate_der(cert)
try:
digest = hashlib.new(algorithm.lower())
except ValueError as exc:
raise ValueError(f"unsupported fingerprint algorithm {algorithm!r}") from exc
digest.update(data)
return digest.hexdigest()
def is_self_signed(cert: x509.Certificate) -> bool:
if cert.subject != cert.issuer:
return False
try:
cert.verify_directly_issued_by(cert)
return True
except Exception: # noqa: BLE001 - invalid signature
return False
def verify_directly_issued_by(cert: x509.Certificate, issuer: x509.Certificate) -> bool:
try:
cert.verify_directly_issued_by(issuer)
return True
except Exception: # noqa: BLE001
return False
def find_issuer(cert: x509.Certificate, candidates: Iterable[x509.Certificate]) -> Optional[x509.Certificate]:
"""Find the best issuer among ``candidates`` (latest expiry wins)."""
if is_self_signed(cert):
return cert
matches = [
candidate
for candidate in candidates
if candidate is not cert
and candidate.subject == cert.issuer
and verify_directly_issued_by(cert, candidate)
]
if not matches:
return None
return max(matches, key=lambda candidate: candidate.not_valid_after_utc)
def chain_for(
cert: x509.Certificate,
lookup: Dict[str, x509.Certificate],
max_depth: int = 20,
) -> List[x509.Certificate]:
"""Follow issuer links in ``lookup`` (issuer_hash -> cert) up to a root."""
chain = [cert]
current = cert
seen = {cert.fingerprint(hashes.SHA256())}
for _ in range(max_depth):
if is_self_signed(current):
break
issuer = lookup.get(name_hash(current.issuer))
if issuer is None or issuer.fingerprint(hashes.SHA256()) in seen:
break
if not verify_directly_issued_by(current, issuer):
break
chain.append(issuer)
seen.add(issuer.fingerprint(hashes.SHA256()))
current = issuer
return chain
def validation_errors(cert: x509.Certificate, chain: Sequence[x509.Certificate]) -> List[Dict[str, Any]]:
"""OpenSSL-backed chain validation, returning error code/message pairs.
XCA validates end-entity certificates with the X.509 strict and policy
flags and lists every error. pyOpenSSL surfaces one error per context, so
we verify the chain once and, for missing issuers, report that explicitly.
"""
errors: List[Dict[str, Any]] = []
if is_self_signed(cert):
return errors
if len(chain) < 2:
return [{"code": 20, "message": "unable to get local issuer certificate"}]
try:
from OpenSSL import crypto
store = crypto.X509Store()
# The root of the supplied chain is trusted; intermediates are
# untrusted inputs, exactly like XCA's trusted/untrusted stacks.
for item in chain[1:]:
converted = crypto.X509.from_cryptography(item)
if is_self_signed(item):
store.add_cert(converted)
else:
store.add_cert(converted)
context = crypto.X509StoreContext(
store,
crypto.X509.from_cryptography(cert),
[crypto.X509.from_cryptography(item) for item in chain[1:]],
)
context.verify_certificate()
except Exception as exc: # noqa: BLE001 - pyOpenSSL raises rich errors
code = getattr(exc, "args", [None])[-1] if getattr(exc, "args", None) else None
if not isinstance(code, int):
code = None
errors.append({"code": code, "message": str(exc)})
return errors
def purposes(cert: x509.Certificate) -> List[str]:
"""A practical purpose list derived from EKU/KU (XCA uses OpenSSL's checks)."""
result: List[str] = []
try:
eku = cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage).value
oids = {oid.dotted_string for oid in eku}
except x509.ExtensionNotFound:
oids = set()
labels = {
"1.3.6.1.5.5.7.3.1": "SSL server",
"1.3.6.1.5.5.7.3.2": "SSL client",
"1.3.6.1.5.5.7.3.3": "Code signing",
"1.3.6.1.5.5.7.3.4": "E-mail protection",
"1.3.6.1.5.5.7.3.8": "Time stamping",
"1.3.6.1.5.5.7.3.9": "OCSP helper",
}
for oid, label in labels.items():
if oid in oids:
result.append(label)
if not oids:
result.append("Any purpose")
return result
def certificate_spec(cert: x509.Certificate) -> Dict[str, Any]:
"""A certificate-creation spec for "similar certificate" / renewal."""
info = certificate_info(cert)
return {
"subject": [entry.as_dict() for entry in entries_from_name(cert.subject)],
"serial_hex": info.serial_hex,
"not_before": info.not_before.isoformat(),
"not_after": info.not_after.isoformat(),
"signature_hash": _signature_hash_name(cert),
"extensions": extension_specs(cert.extensions),
"public_key_der": info.public_key_der.hex(),
}
def signature_hash_name(cert: x509.Certificate) -> Optional[str]:
try:
algorithm = cert.signature_hash_algorithm
except Exception: # noqa: BLE001 - Ed25519
return None
if algorithm is None:
return None
return algorithm.name
# Backwards-compatible private alias used by the app layer.
_signature_hash_name = signature_hash_name
def index_entry(cert: x509.Certificate, *, revoked_at: Optional[datetime] = None) -> str:
"""One line in OpenSSL's ``index.txt`` format."""
now = datetime.now(timezone.utc)
if revoked_at is not None:
flag = "R"
elif cert.not_valid_before_utc <= now <= cert.not_valid_after_utc:
flag = "V"
else:
flag = "E"
expiry = cert.not_valid_after_utc.strftime("%Y%m%d%H%M%SZ")
revoked = revoked_at.astimezone(timezone.utc).strftime("%Y%m%d%H%M%SZ") if revoked_at else ""
subject = cert.subject.rfc4514_string()
return f"{flag}\t{expiry}\t{revoked}\t{serial_to_hex(cert.serial_number)}\tunknown\t{subject}\n"
def openssl_config(cert: x509.Certificate) -> str:
"""An OpenSSL configuration fragment reproducing the certificate."""
lines = [
"[req]",
"distinguished_name = req_distinguished_name",
"prompt = no",
"",
"[req_distinguished_name]",
]
for entry in entries_from_name(cert.subject):
lines.append(f"{entry.name} = {entry.value}")
lines.extend(["", "[v3_ext]"])
config = extension_config_lines(cert.extensions)
if config:
lines.append(config)
return "\n".join(lines) + "\n"
def ics_event(cert: x509.Certificate, name: str = "") -> str:
"""A vCalendar reminder for the certificate's expiry."""
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
start = cert.not_valid_after_utc.strftime("%Y%m%dT%H%M%SZ")
summary = f"Renew certificate: {name or cert.subject.rfc4514_string()}"
description = (
f"The certificate '{name or cert.subject.rfc4514_string()}' "
f"will expire on {cert.not_valid_after_utc.isoformat()}."
)
return "\r\n".join(
[
"BEGIN:VEVENT",
f"UID:{fingerprint(cert, 'sha256')}@kca",
f"DTSTAMP:{stamp}",
f"DTSTART:{start}",
f"SUMMARY:{_ics_escape(summary)}",
f"DESCRIPTION:{_ics_escape(description)}",
"END:VEVENT",
]
)
def _ics_escape(value: str) -> str:
return value.replace("\\", "\\\\").replace(";", "\\;").replace(",", "\\,").replace("\n", "\\n")
def ics_calendar(events: Sequence[str]) -> str:
lines = [
"BEGIN:VCALENDAR",
"VERSION:2.0",
"PRODID:-//KCA//XCA Web//EN",
"CALSCALE:GREGORIAN",
]
for event in events:
lines.extend(event.split("\r\n"))
lines.append("END:VCALENDAR")
return "\r\n".join(lines) + "\r\n"
def certificate_public_key(cert: x509.Certificate) -> PublicKey:
return cert.public_key()
def subject_entries(cert: x509.Certificate) -> List[Dict[str, str]]:
return [entry.as_dict() for entry in entries_from_name(cert.subject)]
def issuer_entries(cert: x509.Certificate) -> List[Dict[str, str]]:
return [entry.as_dict() for entry in entries_from_name(cert.issuer)]
def certificate_extensions(cert: x509.Certificate) -> List[Dict[str, Any]]:
return extension_rows(cert.extensions)
def certificate_extension_specs(cert: x509.Certificate) -> Dict[str, Any]:
return extension_specs(cert.extensions)
def build_self_signed(
*,
subject: x509.Name,
private_key: PrivateKey,
serial: int,
not_before: datetime,
not_after: datetime,
extension_spec: Optional[Dict[str, Any]] = None,
signature_hash: Optional[str] = "sha256",
) -> x509.Certificate:
public_key = private_key.public_key()
extensions = build_extensions(
extension_spec,
public_key=public_key,
subject_name=subject,
issuer_cert=None,
)
return build_certificate(
subject=subject,
public_key=public_key,
issuer_cert=None,
issuer_private_key=private_key,
serial=serial,
not_before=not_before,
not_after=not_after,
extensions=extensions,
signature_hash=signature_hash,
)
def build_signed(
*,
subject: x509.Name,
public_key: PublicKey,
issuer_cert: x509.Certificate,
issuer_private_key: PrivateKey,
serial: int,
not_before: datetime,
not_after: datetime,
extension_spec: Optional[Dict[str, Any]] = None,
signature_hash: Optional[str] = "sha256",
) -> x509.Certificate:
extensions = build_extensions(
extension_spec,
public_key=public_key,
subject_name=subject,
issuer_cert=issuer_cert,
)
return build_certificate(
subject=subject,
public_key=public_key,
issuer_cert=issuer_cert,
issuer_private_key=issuer_private_key,
serial=serial,
not_before=not_before,
not_after=not_after,
extensions=extensions,
signature_hash=signature_hash,
)
__all__ = [
"FAR_FUTURE",
"MAX_SERIAL_BYTES",
"CertificateInfo",
"build_certificate",
"build_self_signed",
"build_signed",
"certificate_der",
"certificate_extension_specs",
"certificate_extensions",
"certificate_info",
"certificate_pem",
"certificate_public_key",
"certificate_spec",
"chain_for",
"find_issuer",
"fingerprint",
"ics_calendar",
"ics_event",
"index_entry",
"is_self_signed",
"issuer_entries",
"normalize_datetime",
"openssl_config",
"parse_certificate",
"parse_serial_hex",
"purposes",
"serial_to_hex",
"signature_algorithm_name",
"signature_hash_name",
"subject_entries",
"validate_chain",
"validation_errors",
"verify_directly_issued_by",
]
@@ -0,0 +1,244 @@
"""Certificate revocation list construction and parsing."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Mapping, Optional
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from cryptography.x509.oid import ExtensionOID
from .certificates import normalize_datetime, signature_algorithm_name
from .extensions import extension_rows, extension_specs
from .keys import PrivateKey, hash_algorithm
#: XCA's revocation reason names, in OpenSSL order.
CRL_REASONS: List[Dict[str, Any]] = [
{"name": "unspecified", "label": "Unspecified", "flag": x509.ReasonFlags.unspecified},
{"name": "keyCompromise", "label": "Key Compromise", "flag": x509.ReasonFlags.key_compromise},
{"name": "CACompromise", "label": "CA Compromise", "flag": x509.ReasonFlags.ca_compromise},
{"name": "affiliationChanged", "label": "Affiliation Changed", "flag": x509.ReasonFlags.affiliation_changed},
{"name": "superseded", "label": "Superseded", "flag": x509.ReasonFlags.superseded},
{"name": "cessationOfOperation", "label": "Cessation Of Operation", "flag": x509.ReasonFlags.cessation_of_operation},
{"name": "certificateHold", "label": "Certificate Hold", "flag": x509.ReasonFlags.certificate_hold},
{"name": "removeFromCRL", "label": "Remove From CRL", "flag": x509.ReasonFlags.remove_from_crl},
{"name": "privilegeWithdrawn", "label": "Privilege Withdrawn", "flag": x509.ReasonFlags.privilege_withdrawn},
{"name": "AACompromise", "label": "AA Compromise", "flag": x509.ReasonFlags.aa_compromise},
]
_REASON_BY_NAME = {item["name"]: item for item in CRL_REASONS}
_REASON_BY_FLAG = {item["flag"]: item for item in CRL_REASONS}
@dataclass(frozen=True)
class CrlInfo:
issuer: x509.Name
last_update: datetime
next_update: Optional[datetime]
crl_number: Optional[int]
revoked_count: int
signature_algorithm: str
fingerprint_sha256: str
def reason_flag(name: str) -> x509.ReasonFlags:
return _REASON_BY_NAME.get(name, _REASON_BY_NAME["unspecified"])["flag"]
def reason_name(flag: x509.ReasonFlags) -> str:
return _REASON_BY_FLAG.get(flag, _REASON_BY_NAME["unspecified"])["name"]
def build_crl(
*,
issuer_cert: x509.Certificate,
issuer_private_key: PrivateKey,
revocations: Iterable[Mapping[str, Any]],
last_update: datetime,
next_update: datetime,
crl_number: Optional[int] = None,
signature_hash: Optional[str] = "sha256",
with_reason: bool = True,
include_aki: bool = True,
include_issuer_alt_name: bool = True,
) -> x509.CertificateRevocationList:
last_update = normalize_datetime(last_update)
next_update = normalize_datetime(next_update)
if next_update <= last_update:
raise ValueError("nextUpdate must be later than lastUpdate")
builder = (
x509.CertificateRevocationListBuilder()
.issuer_name(issuer_cert.subject)
.last_update(last_update)
.next_update(next_update)
)
for item in revocations:
serial = int(str(item.get("serial_hex", "0")).replace(":", ""), 16)
revoked_at = normalize_datetime(item.get("revoked_at") or last_update)
revoked = (
x509.RevokedCertificateBuilder()
.serial_number(serial)
.revocation_date(revoked_at)
)
if with_reason:
revoked = revoked.add_extension(
x509.CRLReason(reason_flag(str(item.get("reason", "unspecified")))),
critical=False,
)
invalidity = item.get("invalidity_date")
if invalidity:
revoked = revoked.add_extension(
x509.InvalidityDate(normalize_datetime(invalidity)), critical=False
)
builder = builder.add_revoked_certificate(revoked.build())
if crl_number is not None:
builder = builder.add_extension(x509.CRLNumber(int(crl_number)), critical=False)
if include_aki:
try:
ski = issuer_cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier).value
aki = x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(ski)
except x509.ExtensionNotFound:
aki = x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_cert.public_key())
builder = builder.add_extension(aki, critical=False)
if include_issuer_alt_name:
try:
san = issuer_cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
builder = builder.add_extension(x509.IssuerAlternativeName(list(san)), critical=False)
except x509.ExtensionNotFound:
pass
algorithm = hash_algorithm(signature_hash, issuer_private_key)
return builder.sign(private_key=issuer_private_key, algorithm=algorithm)
def parse_crl(data: bytes) -> x509.CertificateRevocationList:
return x509.load_der_x509_crl(data)
def crl_der(crl: x509.CertificateRevocationList) -> bytes:
return crl.public_bytes(serialization.Encoding.DER)
def crl_pem(crl: x509.CertificateRevocationList) -> str:
return crl.public_bytes(serialization.Encoding.PEM).decode("ascii")
def crl_number(crl: x509.CertificateRevocationList) -> Optional[int]:
try:
return crl.extensions.get_extension_for_class(x509.CRLNumber).value.crl_number
except x509.ExtensionNotFound:
return None
def crl_info(crl: x509.CertificateRevocationList) -> CrlInfo:
return CrlInfo(
issuer=crl.issuer,
last_update=normalize_datetime(crl.last_update_utc),
next_update=normalize_datetime(crl.next_update_utc) if crl.next_update_utc else None,
crl_number=crl_number(crl),
revoked_count=len(list(crl)),
signature_algorithm=signature_algorithm_name(crl),
fingerprint_sha256=hashlib.sha256(crl_der(crl)).hexdigest(),
)
def crl_revocations(crl: x509.CertificateRevocationList) -> List[Dict[str, Any]]:
rows = []
for revoked in crl:
reason = None
invalidity = None
try:
reason = reason_name(
revoked.extensions.get_extension_for_class(x509.CRLReason).value.reason
)
except x509.ExtensionNotFound:
pass
try:
invalidity = normalize_datetime(
revoked.extensions.get_extension_for_class(x509.InvalidityDate).value.invalidity_date
).isoformat()
except x509.ExtensionNotFound:
pass
rows.append(
{
"serial_hex": format(revoked.serial_number, "x"),
"revoked_at": normalize_datetime(revoked.revocation_date_utc).isoformat(),
"invalidity_date": invalidity,
"reason": reason,
}
)
return rows
def verify_crl(crl: x509.CertificateRevocationList, issuer_cert: x509.Certificate) -> bool:
if crl.issuer != issuer_cert.subject:
return False
try:
return bool(crl.is_signature_valid(issuer_cert.public_key()))
except Exception: # noqa: BLE001
return False
def crl_extensions(crl: x509.CertificateRevocationList) -> List[Dict[str, Any]]:
return extension_rows(crl.extensions)
def crl_spec(crl: x509.CertificateRevocationList) -> Dict[str, Any]:
return {
"issuer": crl.issuer.rfc4514_string(),
"last_update": normalize_datetime(crl.last_update_utc).isoformat(),
"next_update": normalize_datetime(crl.next_update_utc).isoformat() if crl.next_update_utc else None,
"crl_number": crl_number(crl),
"revocations": crl_revocations(crl),
"extensions": extension_specs(crl.extensions),
}
def ics_event(crl: x509.CertificateRevocationList, name: str = "") -> str:
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
next_update = crl.next_update_utc
if next_update is None:
return ""
start = next_update.strftime("%Y%m%dT%H%M%SZ")
summary = f"Renew CRL: {name or crl.issuer.rfc4514_string()}"
description = f"The CRL '{name or crl.issuer.rfc4514_string()}' expires on {next_update.isoformat()}."
return "\r\n".join(
[
"BEGIN:VEVENT",
f"UID:{hashlib.sha256(crl_der(crl)).hexdigest()}@kca",
f"DTSTAMP:{stamp}",
f"DTSTART:{start}",
f"SUMMARY:{summary.replace(',', '\\,')}",
f"DESCRIPTION:{description.replace(',', '\\,')}",
"END:VEVENT",
]
)
def reason_options() -> List[Dict[str, str]]:
return [{"name": item["name"], "label": item["label"]} for item in CRL_REASONS]
__all__ = [
"CRL_REASONS",
"CrlInfo",
"build_crl",
"crl_der",
"crl_extensions",
"crl_info",
"crl_number",
"crl_pem",
"crl_revocations",
"crl_spec",
"ics_event",
"parse_crl",
"reason_flag",
"reason_name",
"reason_options",
"verify_crl",
]
@@ -0,0 +1,83 @@
"""Small AES-256-GCM envelope used for private keys at rest."""
from __future__ import annotations
import base64
import os
from dataclasses import dataclass
from typing import Optional
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
NONCE_BYTES = 12
KEY_BYTES = 32
class DecryptionError(ValueError):
"""The ciphertext could not be authenticated with the configured key."""
@dataclass(frozen=True)
class EncryptedBlob:
ciphertext: bytes
nonce: bytes
def decode_master_key(value: str) -> bytes:
"""Decode a base64 (or hex) 32-byte master key from the environment."""
text = (value or "").strip()
if not text:
raise ValueError("KCA_MASTER_KEY is empty")
try:
raw = base64.b64decode(text, validate=True)
except Exception: # noqa: BLE001
try:
raw = bytes.fromhex(text)
except ValueError as exc:
raise ValueError("KCA_MASTER_KEY must be base64 or hex") from exc
if len(raw) != KEY_BYTES:
raise ValueError(f"KCA_MASTER_KEY must be {KEY_BYTES} bytes")
return raw
def generate_master_key() -> str:
return base64.b64encode(os.urandom(KEY_BYTES)).decode("ascii")
def encrypt_blob(plaintext: bytes, *, master_key: bytes, aad: bytes = b"") -> EncryptedBlob:
if len(master_key) != KEY_BYTES:
raise ValueError(f"master key must be {KEY_BYTES} bytes")
nonce = os.urandom(NONCE_BYTES)
ciphertext = AESGCM(master_key).encrypt(nonce, plaintext, aad)
return EncryptedBlob(ciphertext=ciphertext, nonce=nonce)
def decrypt_blob(blob: EncryptedBlob, *, master_key: bytes, aad: bytes = b"") -> bytes:
if len(master_key) != KEY_BYTES:
raise ValueError(f"master key must be {KEY_BYTES} bytes")
try:
return AESGCM(master_key).decrypt(blob.nonce, blob.ciphertext, aad)
except InvalidTag as exc:
raise DecryptionError("private key authentication failed") from exc
def encrypt_private_key(der: bytes, key_id: str, master_key: bytes) -> EncryptedBlob:
return encrypt_blob(der, master_key=master_key, aad=key_id.encode("utf-8"))
def decrypt_private_key(blob: EncryptedBlob, key_id: str, master_key: bytes) -> bytes:
return decrypt_blob(blob, master_key=master_key, aad=key_id.encode("utf-8"))
__all__ = [
"DecryptionError",
"EncryptedBlob",
"KEY_BYTES",
"NONCE_BYTES",
"decode_master_key",
"decrypt_blob",
"decrypt_private_key",
"encrypt_blob",
"encrypt_private_key",
"generate_master_key",
]
@@ -0,0 +1,324 @@
"""Export formats for certificates, keys, requests, CRLs and templates."""
from __future__ import annotations
import base64
import hashlib
import json
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Optional, Sequence
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.serialization import (
BestAvailableEncryption,
Encoding,
NoEncryption,
pkcs7,
pkcs12,
)
from .certificates import certificate_der, certificate_pem, index_entry, ics_calendar, ics_event, openssl_config
from .crls import crl_der, crl_pem, ics_event as crl_ics_event
from .keys import (
PrivateKey,
jwk_private,
jwk_public,
private_der_bytes,
private_pem,
public_pem,
pvk_encode_private,
ssh_private,
ssh_public,
)
from .requests import csr_der, csr_pem, openssl_config as csr_openssl_config
from .templates import template_to_xca_data
OPENVPN_TA_BEGIN = "-----BEGIN OpenVPN Static key V1-----"
OPENVPN_TA_END = "-----END OpenVPN Static key V1-----"
@dataclass(frozen=True)
class ExportResult:
filename: str
content_type: str
data: bytes
def _safe_name(name: str, fallback: str) -> str:
cleaned = "".join(ch if ch.isalnum() or ch in "-_." else "_" for ch in (name or "")).strip("._")
return cleaned or fallback
def _b64u(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
def _password_encryption(password: Optional[str]) -> serialization.KeySerializationEncryption:
if password:
return BestAvailableEncryption(password.encode())
return NoEncryption()
def _pem_chain(cert: x509.Certificate, chain: Sequence[x509.Certificate]) -> bytes:
parts = [certificate_pem(cert)]
for item in chain:
if item is cert:
continue
parts.append(certificate_pem(item))
return "".join(parts).encode("ascii")
def export_certificate(
cert: x509.Certificate,
export_format: str,
*,
name: str = "",
chain: Sequence[x509.Certificate] = (),
private_key: Optional[PrivateKey] = None,
password: Optional[str] = None,
tls_auth_key: Optional[str] = None,
comment: str = "",
) -> ExportResult:
fmt = (export_format or "pem").lower()
stem = _safe_name(name, "certificate")
if fmt == "pem":
return ExportResult(f"{stem}.crt", "application/x-pem-file", certificate_pem(cert).encode("ascii"))
if fmt in ("pem-chain", "chain"):
return ExportResult(f"{stem}.pem", "application/x-pem-file", _pem_chain(cert, chain))
if fmt == "pem-key":
if private_key is None:
raise ValueError("this certificate has no private key")
data = certificate_pem(cert).encode("ascii") + private_pem(private_key)
return ExportResult(f"{stem}.pem", "application/x-pem-file", data)
if fmt == "pem-pkcs8":
if private_key is None:
raise ValueError("this certificate has no private key")
if not password:
raise ValueError("a password is required for encrypted PKCS#8 export")
data = certificate_pem(cert).encode("ascii") + private_pem(private_key, password=password)
return ExportResult(f"{stem}.pem", "application/x-pem-file", data)
if fmt in ("pkcs7", "pkcs7-chain", "pkcs7-multi"):
certs = [cert]
if fmt != "pkcs7":
certs.extend(item for item in chain if item is not cert)
data = pkcs7.serialize_certificates(certs, Encoding.DER)
return ExportResult(f"{stem}.p7b", "application/pkcs7-mime", data)
if fmt == "der":
return ExportResult(f"{stem}.cer", "application/pkix-cert", certificate_der(cert))
if fmt in ("pkcs12", "pfx", "pkcs12-chain"):
if private_key is None:
raise ValueError("PKCS#12 export requires the private key")
cas = [item for item in chain if item is not cert] if fmt == "pkcs12-chain" else []
data = pkcs12.serialize_key_and_certificates(
name=stem.encode(),
key=private_key,
cert=cert,
cas=cas or None,
encryption_algorithm=_password_encryption(password),
)
return ExportResult(f"{stem}.pfx", "application/x-pkcs12", data)
if fmt == "openvpn":
if private_key is None:
raise ValueError("OpenVPN export requires the private key")
ca_cert = next((item for item in reversed(chain) if item is not cert), None)
ca_pem = certificate_pem(ca_cert) if ca_cert is not None else certificate_pem(cert)
key_pem = private_pem(private_key).decode("ascii")
text = (
f"<ca>\n{ca_pem}</ca>\n"
f"<cert>\n{certificate_pem(cert)}</cert>\n"
f"<key>\n{key_pem}</key>\n"
)
return ExportResult(f"{stem}.ovpn", "text/plain", text.encode("utf-8"))
if fmt in ("openssl-config", "config"):
return ExportResult(f"{stem}.conf", "text/plain", openssl_config(cert).encode("utf-8"))
if fmt == "index":
return ExportResult(f"{stem}.txt", "text/plain", index_entry(cert).encode("utf-8"))
if fmt == "vcalendar":
return ExportResult(f"{stem}.ics", "text/calendar", ics_calendar([ics_event(cert, name)]).encode("utf-8"))
if fmt in ("jwk", "jwk-public", "jwk-chain"):
jwk = jwk_public(cert.public_key(), kid=name or None)
der = certificate_der(cert)
jwk["x5t"] = _b64u(hashlib.sha1(der).digest())
jwk["x5t#256"] = _b64u(hashlib.sha256(der).digest())
if fmt == "jwk-chain":
jwk["x5c"] = [base64.b64encode(certificate_der(item)).decode("ascii") for item in [cert, *chain]]
return ExportResult(f"{stem}.jwk", "application/json", json.dumps(jwk, indent=2).encode("utf-8"))
if fmt == "tls-auth":
if not tls_auth_key:
raise ValueError("no OpenVPN tls-auth key is stored for this CA")
return ExportResult(f"{stem}.key", "text/plain", tls_auth_key.encode("ascii"))
raise ValueError(f"unsupported certificate export format {export_format!r}")
def export_key(
key: PrivateKey | Any,
export_format: str,
*,
name: str = "",
password: Optional[str] = None,
is_private: bool = True,
) -> ExportResult:
fmt = (export_format or "pem-public").lower()
stem = _safe_name(name, "key")
if fmt == "pem-public":
return ExportResult(f"{stem}.pem", "application/x-pem-file", public_pem(key).encode("ascii"))
if fmt in ("pem-private", "pem"):
if not is_private:
raise ValueError("this is a public key")
return ExportResult(f"{stem}.pem", "application/x-pem-file", private_pem(key))
if fmt == "pem-encrypted":
if not is_private:
raise ValueError("this is a public key")
if not password:
raise ValueError("a password is required")
return ExportResult(f"{stem}.pem", "application/x-pem-file", private_pem(key, password=password, traditional=True))
if fmt == "der-public":
from .keys import public_der_bytes
return ExportResult(f"{stem}.der", "application/pkix-key", public_der_bytes(key))
if fmt == "der-private":
if not is_private:
raise ValueError("this is a public key")
return ExportResult(f"{stem}.der", "application/pkcs8", private_der_bytes(key))
if fmt in ("pkcs8", "pkcs8-private"):
if not is_private:
raise ValueError("this is a public key")
return ExportResult(f"{stem}.pk8", "application/x-pem-file", private_pem(key))
if fmt == "pkcs8-encrypted":
if not is_private:
raise ValueError("this is a public key")
if not password:
raise ValueError("a password is required")
return ExportResult(f"{stem}.pk8", "application/x-pem-file", private_pem(key, password=password))
if fmt == "ssh-public":
return ExportResult(f"{stem}.pub", "text/plain", (ssh_public(key, name) + "\n").encode("utf-8"))
if fmt == "ssh-private":
if not is_private:
raise ValueError("this is a public key")
return ExportResult(f"{stem}.priv", "text/plain", ssh_private(key, password=password))
if fmt in ("jwk", "jwk-public"):
return ExportResult(f"{stem}.jwk", "application/json", json.dumps(jwk_public(key, kid=name or None), indent=2).encode("utf-8"))
if fmt == "jwk-private":
if not is_private:
raise ValueError("this is a public key")
return ExportResult(f"{stem}.jwk", "application/json", json.dumps(jwk_private(key, kid=name or None), indent=2).encode("utf-8"))
if fmt == "pvk":
if not is_private:
raise ValueError("this is a public key")
return ExportResult(f"{stem}.pvk", "application/octet-stream", pvk_encode_private(key))
raise ValueError(f"unsupported key export format {export_format!r}")
def export_request(
csr: x509.CertificateSigningRequest,
export_format: str,
*,
name: str = "",
) -> ExportResult:
fmt = (export_format or "pem").lower()
stem = _safe_name(name, "request")
if fmt == "pem":
return ExportResult(f"{stem}.csr", "application/x-pem-file", csr_pem(csr).encode("ascii"))
if fmt == "der":
return ExportResult(f"{stem}.der", "application/pkcs10", csr_der(csr))
if fmt in ("openssl-config", "config"):
return ExportResult(f"{stem}.conf", "text/plain", csr_openssl_config(csr).encode("utf-8"))
raise ValueError(f"unsupported request export format {export_format!r}")
def export_crl(crl: x509.CertificateRevocationList, export_format: str, *, name: str = "") -> ExportResult:
fmt = (export_format or "pem").lower()
stem = _safe_name(name, "crl")
if fmt == "pem":
return ExportResult(f"{stem}.crl", "application/x-pem-file", crl_pem(crl).encode("ascii"))
if fmt == "der":
return ExportResult(f"{stem}.crl", "application/pkix-crl", crl_der(crl))
if fmt == "vcalendar":
return ExportResult(f"{stem}.ics", "text/calendar", ics_calendar([crl_ics_event(crl, name)]).encode("utf-8"))
raise ValueError(f"unsupported CRL export format {export_format!r}")
def export_template(template: Dict[str, Any], export_format: str = "xca", *, name: str = "") -> ExportResult:
fmt = (export_format or "xca").lower()
stem = _safe_name(name or str(template.get("name", "")), "template")
if fmt in ("xca", "pem"):
return ExportResult(f"{stem}.xca", "application/x-pem-file", template_to_xca_data(template))
if fmt == "json":
return ExportResult(f"{stem}.json", "application/json", json.dumps(template, indent=2).encode("utf-8"))
raise ValueError(f"unsupported template export format {export_format!r}")
def certificate_format_options() -> List[Dict[str, str]]:
return [
{"id": "pem", "label": "PEM", "extension": "crt"},
{"id": "pem-chain", "label": "PEM chain", "extension": "pem"},
{"id": "pem-key", "label": "PEM + key", "extension": "pem"},
{"id": "pem-pkcs8", "label": "PEM + encrypted PKCS#8", "extension": "pem"},
{"id": "pkcs7", "label": "PKCS#7", "extension": "p7b"},
{"id": "pkcs7-chain", "label": "PKCS#7 chain", "extension": "p7b"},
{"id": "der", "label": "DER", "extension": "cer"},
{"id": "pkcs12", "label": "PKCS#12", "extension": "pfx"},
{"id": "pkcs12-chain", "label": "PKCS#12 chain", "extension": "pfx"},
{"id": "openvpn", "label": "OpenVPN", "extension": "ovpn"},
{"id": "openssl-config", "label": "OpenSSL config", "extension": "conf"},
{"id": "index", "label": "Certificate index file", "extension": "txt"},
{"id": "vcalendar", "label": "vCalendar", "extension": "ics"},
{"id": "jwk", "label": "JSON Web Key", "extension": "jwk"},
{"id": "jwk-chain", "label": "JSON Web Key chain", "extension": "jwk"},
{"id": "tls-auth", "label": "OpenVPN tls-auth key", "extension": "key"},
]
def key_format_options() -> List[Dict[str, str]]:
return [
{"id": "pem-public", "label": "PEM public", "extension": "pem"},
{"id": "pem-private", "label": "PEM private", "extension": "pem"},
{"id": "pem-encrypted", "label": "PEM encrypted", "extension": "pem"},
{"id": "pkcs8", "label": "PKCS#8", "extension": "pk8"},
{"id": "pkcs8-encrypted", "label": "PKCS#8 encrypted", "extension": "pk8"},
{"id": "der-public", "label": "DER public", "extension": "der"},
{"id": "der-private", "label": "DER private", "extension": "der"},
{"id": "ssh-public", "label": "SSH2 public", "extension": "pub"},
{"id": "ssh-private", "label": "SSH2 private", "extension": "priv"},
{"id": "jwk-public", "label": "JSON Web Key public", "extension": "jwk"},
{"id": "jwk-private", "label": "JSON Web Key private", "extension": "jwk"},
{"id": "pvk", "label": "PVK private", "extension": "pvk"},
]
def request_format_options() -> List[Dict[str, str]]:
return [
{"id": "pem", "label": "PEM", "extension": "csr"},
{"id": "der", "label": "DER", "extension": "der"},
{"id": "openssl-config", "label": "OpenSSL config", "extension": "conf"},
]
def crl_format_options() -> List[Dict[str, str]]:
return [
{"id": "pem", "label": "PEM", "extension": "crl"},
{"id": "der", "label": "DER", "extension": "crl"},
{"id": "vcalendar", "label": "vCalendar", "extension": "ics"},
]
def template_format_options() -> List[Dict[str, str]]:
return [
{"id": "xca", "label": "XCA template", "extension": "xca"},
{"id": "json", "label": "JSON", "extension": "json"},
]
__all__ = [
"ExportResult",
"certificate_format_options",
"crl_format_options",
"export_certificate",
"export_crl",
"export_key",
"export_request",
"export_template",
"key_format_options",
"request_format_options",
"template_format_options",
]
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,239 @@
"""Import parsing for PEM bundles, DER items, PKCS#7, PKCS#12 and SSH keys."""
from __future__ import annotations
import base64
import re
from dataclasses import dataclass, field
from typing import Any, Dict, Iterable, List, Optional, Sequence
from cryptography import x509
from cryptography.hazmat.primitives.serialization import pkcs7, pkcs12
from .certificates import certificate_info
from .crls import crl_info
from .keys import PrivateKey, PublicKey, is_private_key, key_info, load_key_bytes
from .names import entries_from_name
from .requests import request_info
from .templates import template_from_xca_data
_PEM_BLOCK = re.compile(
r"-----BEGIN ([^-]+)-----\s*(.*?)\s*-----END \1-----",
re.DOTALL,
)
@dataclass
class ParsedItem:
kind: str
name: str
summary: str
obj: Any
is_private: bool = False
warnings: List[str] = field(default_factory=list)
extra: Dict[str, Any] = field(default_factory=dict)
@property
def fingerprint(self) -> Optional[str]:
if self.kind == "certificate":
return certificate_info(self.obj).fingerprint_sha256
if self.kind == "key":
return key_info(self.obj).public_sha256
if self.kind == "request":
return request_info(self.obj).fingerprint_sha256
if self.kind == "crl":
return crl_info(self.obj).fingerprint_sha256
return None
def _guess_name(obj: Any, filename: str = "") -> str:
stem = filename.rsplit("/", 1)[-1]
if "." in stem:
stem = stem.rsplit(".", 1)[0]
if isinstance(obj, x509.Certificate):
names = [str(attr.value) for attr in obj.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)]
return names[0] if names else (stem or "certificate")
if isinstance(obj, x509.CertificateSigningRequest):
names = [str(attr.value) for attr in obj.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)]
return names[0] if names else (stem or "request")
if isinstance(obj, x509.CertificateRevocationList):
return stem or "crl"
if isinstance(obj, dict):
return str(obj.get("name") or stem or "template")
return stem or "key"
def _summary(obj: Any) -> str:
info: Any
if isinstance(obj, x509.Certificate):
info = certificate_info(obj)
return f"{info.subject.rfc4514_string()} · {info.serial_hex}"
if isinstance(obj, x509.CertificateSigningRequest):
info = request_info(obj)
return info.subject.rfc4514_string()
if isinstance(obj, x509.CertificateRevocationList):
info = crl_info(obj)
return f"{info.issuer.rfc4514_string()} · {info.revoked_count} revoked"
if isinstance(obj, dict):
return str(obj.get("comment") or "XCA template")
try:
info = key_info(obj)
return f"{info.key_type} {info.key_size}" + (f" {info.curve}" if info.curve else "")
except Exception: # noqa: BLE001
return ""
def _key_item(key: Any, filename: str, name: Optional[str] = None) -> ParsedItem:
private = is_private_key(key)
return ParsedItem(
kind="key",
name=name or _guess_name(key, filename),
summary=_summary(key),
obj=key,
is_private=private,
)
def _cert_item(cert: x509.Certificate, filename: str, name: Optional[str] = None) -> ParsedItem:
return ParsedItem(
kind="certificate",
name=name or _guess_name(cert, filename),
summary=_summary(cert),
obj=cert,
)
def _csr_item(csr: x509.CertificateSigningRequest, filename: str, name: Optional[str] = None) -> ParsedItem:
return ParsedItem(
kind="request",
name=name or _guess_name(csr, filename),
summary=_summary(csr),
obj=csr,
)
def _crl_item(crl: x509.CertificateRevocationList, filename: str, name: Optional[str] = None) -> ParsedItem:
return ParsedItem(
kind="crl",
name=name or _guess_name(crl, filename),
summary=_summary(crl),
obj=crl,
)
def _template_item(template: Dict[str, Any], filename: str) -> ParsedItem:
return ParsedItem(
kind="template",
name=str(template.get("name") or _guess_name(template, filename)),
summary=_summary(template),
obj=template,
)
def parse_pkcs12(data: bytes, filename: str = "", password: Optional[str] = None) -> List[ParsedItem]:
try:
key, cert, cas = pkcs12.load_key_and_certificates(data, password.encode() if password else None)
except Exception as exc: # noqa: BLE001
raise ValueError("unable to read the PKCS#12 file (wrong password or unsupported encryption)") from exc
items: List[ParsedItem] = []
name = _guess_name(cert, filename) if cert is not None else _guess_name(None, filename)
if cert is not None:
items.append(_cert_item(cert, filename, name))
if key is not None:
items.append(_key_item(key, filename, f"{name}_key"))
for index, extra in enumerate(cas or []):
items.append(_cert_item(extra, filename, f"{name}_ca_{index}"))
if not items:
raise ValueError("the PKCS#12 file contains no keys or certificates")
return items
def parse_pkcs7(data: bytes, filename: str = "") -> List[ParsedItem]:
certs: Sequence[x509.Certificate] = []
try:
certs = pkcs7.load_der_pkcs7_certificates(data)
except Exception: # noqa: BLE001
try:
certs = pkcs7.load_pem_pkcs7_certificates(data)
except Exception as exc: # noqa: BLE001
raise ValueError("unable to read the PKCS#7 file") from exc
if not certs:
raise ValueError("the PKCS#7 file contains no certificates")
return [_cert_item(cert, filename, f"{_guess_name(cert, filename)}_{index}" if len(certs) > 1 else None) for index, cert in enumerate(certs)]
def parse_pem_bundle(data: bytes, filename: str = "", password: Optional[str] = None) -> List[ParsedItem]:
text = data.decode("utf-8", "replace")
blocks = list(_PEM_BLOCK.finditer(text))
if not blocks:
return []
items: List[ParsedItem] = []
for match in blocks:
label = match.group(1).strip().upper()
block = match.group(0).encode("utf-8")
try:
if label == "CERTIFICATE":
items.append(_cert_item(x509.load_pem_x509_certificate(block), filename))
elif label in ("X509 CRL", "CRL"):
items.append(_crl_item(x509.load_pem_x509_crl(block), filename))
elif label in ("CERTIFICATE REQUEST", "NEW CERTIFICATE REQUEST"):
items.append(_csr_item(x509.load_pem_x509_csr(block), filename))
elif label in ("PKCS7", "PKCS #7"):
items.extend(parse_pkcs7(block, filename))
elif label == "XCA TEMPLATE":
items.append(_template_item(template_from_xca_data(block), filename))
elif "PRIVATE KEY" in label or "PUBLIC KEY" in label or label in ("OPENSSH PRIVATE KEY", "SSH2 PUBLIC KEY"):
key, private = load_key_bytes(block, password)
items.append(_key_item(key, filename))
else:
# Unknown PEM label: try every parser before giving up.
items.extend(_try_all(block, filename, password))
except Exception as exc: # noqa: BLE001
raise ValueError(f"unable to import the {label} block: {exc}") from exc
return items
def _try_all(data: bytes, filename: str, password: Optional[str]) -> List[ParsedItem]:
for parser in (
lambda: [_cert_item(x509.load_der_x509_certificate(data), filename)],
lambda: [_csr_item(x509.load_der_x509_csr(data), filename)],
lambda: [_crl_item(x509.load_der_x509_crl(data), filename)],
lambda: parse_pkcs12(data, filename, password),
lambda: parse_pkcs7(data, filename),
lambda: [_key_item(load_key_bytes(data, password)[0], filename)],
):
try:
return parser()
except Exception: # noqa: BLE001
continue
raise ValueError("unrecognized data")
def parse_file(data: bytes, filename: str = "", password: Optional[str] = None) -> List[ParsedItem]:
"""Parse any supported file, detecting PEM vs DER automatically."""
if not data:
raise ValueError("the file is empty")
pem_items = parse_pem_bundle(data, filename, password)
if pem_items:
return pem_items
return _try_all(data, filename, password)
def parse_text(text: str, filename: str = "", password: Optional[str] = None) -> List[ParsedItem]:
return parse_file(text.encode("utf-8"), filename, password)
def parse_xca_database(data: bytes, filename: str = "", password: Optional[str] = None) -> List[ParsedItem]:
from .xca_db import import_xca_database
return import_xca_database(data, filename, password)
__all__ = [
"ParsedItem",
"parse_file",
"parse_pem_bundle",
"parse_pkcs7",
"parse_pkcs12",
"parse_text",
"parse_xca_database",
]
@@ -0,0 +1,543 @@
"""Key generation, parsing and serialization.
All key material handled here is in memory. The application layer is
responsible for encrypting private keys before they touch the database.
"""
from __future__ import annotations
import base64
import hashlib
import io
import struct
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import dsa, ec, ed25519, rsa
from cryptography.hazmat.primitives.serialization import (
Encoding,
NoEncryption,
PrivateFormat,
PublicFormat,
)
PrivateKey = (
rsa.RSAPrivateKey
| dsa.DSAPrivateKey
| ec.EllipticCurvePrivateKey
| ed25519.Ed25519PrivateKey
)
PublicKey = (
rsa.RSAPublicKey
| dsa.DSAPublicKey
| ec.EllipticCurvePublicKey
| ed25519.Ed25519PublicKey
)
#: Curves offered by the UI. ``cryptography`` supports all of them on every
#: platform we target.
EC_CURVES: Dict[str, ec.EllipticCurve] = {
"secp192r1": ec.SECP192R1(),
"secp224r1": ec.SECP224R1(),
"secp256r1": ec.SECP256R1(),
"secp384r1": ec.SECP384R1(),
"secp521r1": ec.SECP521R1(),
"secp256k1": ec.SECP256K1(),
"brainpoolP256r1": ec.BrainpoolP256R1(),
"brainpoolP384r1": ec.BrainpoolP384R1(),
"brainpoolP512r1": ec.BrainpoolP512R1(),
}
_CURVE_NAMES = {curve.name: name for name, curve in EC_CURVES.items()}
KEY_TYPES = ("rsa", "dsa", "ec", "ed25519")
DEFAULT_KEY_SIZE = 2048
MAX_KEY_SIZE = 8192
MIN_KEY_SIZE = 512
#: Hash algorithms selectable for signatures. Ed25519 has no separate hash.
HASH_ALGORITHMS: Dict[str, Optional[hashes.HashAlgorithm]] = {
"sha1": hashes.SHA1(),
"sha224": hashes.SHA224(),
"sha256": hashes.SHA256(),
"sha384": hashes.SHA384(),
"sha512": hashes.SHA512(),
}
@dataclass(frozen=True)
class KeyInfo:
key_type: str
key_size: int
curve: Optional[str]
public_der: bytes
private_der: Optional[bytes]
is_private: bool
@property
def public_sha256(self) -> str:
return hashlib.sha256(self.public_der).hexdigest()
@property
def fingerprint_sha256(self) -> str:
return self.public_sha256
def _require_private(key: Any) -> PrivateKey:
if not isinstance(
key, (rsa.RSAPrivateKey, dsa.DSAPrivateKey, ec.EllipticCurvePrivateKey, ed25519.Ed25519PrivateKey)
):
raise ValueError("a private key is required")
return key
def _public_key(key: PrivateKey | PublicKey) -> PublicKey:
if isinstance(key, (rsa.RSAPublicKey, dsa.DSAPublicKey, ec.EllipticCurvePublicKey, ed25519.Ed25519PublicKey)):
return key
return key.public_key()
def generate_key(key_type: str, size: int = DEFAULT_KEY_SIZE, curve: Optional[str] = None) -> PrivateKey:
"""Generate a private key of the requested type."""
key_type = (key_type or "").lower()
if key_type == "rsa":
if not (MIN_KEY_SIZE <= int(size) <= MAX_KEY_SIZE):
raise ValueError(f"RSA key size must be between {MIN_KEY_SIZE} and {MAX_KEY_SIZE} bits")
return rsa.generate_private_key(public_exponent=65537, key_size=int(size))
if key_type == "dsa":
if not (MIN_KEY_SIZE <= int(size) <= 3072):
raise ValueError("DSA key size must be between 512 and 3072 bits")
return dsa.generate_private_key(key_size=int(size))
if key_type == "ec":
curve_name = curve or "secp256r1"
if curve_name not in EC_CURVES:
raise ValueError(f"unsupported EC curve {curve_name!r}")
return ec.generate_private_key(EC_CURVES[curve_name])
if key_type == "ed25519":
return ed25519.Ed25519PrivateKey.generate()
raise ValueError(f"unsupported key type {key_type!r}")
def key_type_name(key: PrivateKey | PublicKey) -> str:
if isinstance(key, (rsa.RSAPrivateKey, rsa.RSAPublicKey)):
return "RSA"
if isinstance(key, (dsa.DSAPrivateKey, dsa.DSAPublicKey)):
return "DSA"
if isinstance(key, (ec.EllipticCurvePrivateKey, ec.EllipticCurvePublicKey)):
return "EC"
if isinstance(key, (ed25519.Ed25519PrivateKey, ed25519.Ed25519PublicKey)):
return "ED25519"
raise ValueError("unsupported key type")
def key_size(key: PrivateKey | PublicKey) -> int:
if isinstance(key, (rsa.RSAPrivateKey, rsa.RSAPublicKey, dsa.DSAPrivateKey, dsa.DSAPublicKey)):
return key.key_size
if isinstance(key, (ec.EllipticCurvePrivateKey, ec.EllipticCurvePublicKey)):
return key.curve.key_size
if isinstance(key, (ed25519.Ed25519PrivateKey, ed25519.Ed25519PublicKey)):
return 256
raise ValueError("unsupported key type")
def curve_name(key: PrivateKey | PublicKey) -> Optional[str]:
if isinstance(key, (ec.EllipticCurvePrivateKey, ec.EllipticCurvePublicKey)):
return _CURVE_NAMES.get(key.curve.name, key.curve.name)
return None
def is_private_key(key: Any) -> bool:
return isinstance(
key, (rsa.RSAPrivateKey, dsa.DSAPrivateKey, ec.EllipticCurvePrivateKey, ed25519.Ed25519PrivateKey)
)
def public_der(key: PrivateKey | PublicKey) -> bytes:
return _public_key(key).public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
def private_der(key: PrivateKey) -> bytes:
return _require_private(key).private_bytes(Encoding.DER, PrivateFormat.PKCS8, NoEncryption())
def key_info(key: PrivateKey | PublicKey) -> KeyInfo:
private = is_private_key(key)
return KeyInfo(
key_type=key_type_name(key),
key_size=key_size(key),
curve=curve_name(key),
public_der=public_der(key),
private_der=private_der(key) if private else None,
is_private=private,
)
def possible_hashes(key: PrivateKey | PublicKey) -> List[str]:
if isinstance(key, (ed25519.Ed25519PrivateKey, ed25519.Ed25519PublicKey)):
return []
if isinstance(key, (dsa.DSAPrivateKey, dsa.DSAPublicKey)):
return ["sha1", "sha224", "sha256"]
if isinstance(key, (ec.EllipticCurvePrivateKey, ec.EllipticCurvePublicKey)):
# Every supported curve is large enough for SHA-512 in practice.
return ["sha1", "sha224", "sha256", "sha384", "sha512"]
return ["sha1", "sha224", "sha256", "sha384", "sha512"]
def hash_algorithm(name: Optional[str], key: PrivateKey | PublicKey) -> Optional[hashes.HashAlgorithm]:
"""The signature hash for ``key``; ``None`` means the key's built-in hash."""
if isinstance(key, (ed25519.Ed25519PrivateKey, ed25519.Ed25519PublicKey)):
return None
name = (name or "sha256").lower()
if name not in HASH_ALGORITHMS:
raise ValueError(f"unsupported hash algorithm {name!r}")
return HASH_ALGORITHMS[name]
# ---------------------------------------------------------------------------
# Serialization
# ---------------------------------------------------------------------------
def public_pem(key: PrivateKey | PublicKey) -> str:
return _public_key(key).public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo).decode("ascii")
def public_der_bytes(key: PrivateKey | PublicKey) -> bytes:
return public_der(key)
def private_pem(key: PrivateKey, password: Optional[str] = None, traditional: bool = False) -> bytes:
key = _require_private(key)
encryption: serialization.KeySerializationEncryption
encryption = NoEncryption() if not password else serialization.BestAvailableEncryption(password.encode())
fmt = PrivateFormat.TraditionalOpenSSL if traditional else PrivateFormat.PKCS8
if traditional and isinstance(key, ed25519.Ed25519PrivateKey):
raise ValueError("Ed25519 keys cannot be exported in traditional PEM format")
return key.private_bytes(Encoding.PEM, fmt, encryption)
def private_der_bytes(key: PrivateKey, password: Optional[str] = None) -> bytes:
key = _require_private(key)
encryption: serialization.KeySerializationEncryption
encryption = NoEncryption() if not password else serialization.BestAvailableEncryption(password.encode())
return key.private_bytes(Encoding.DER, PrivateFormat.PKCS8, encryption)
def ssh_public(key: PrivateKey | PublicKey, comment: str = "") -> str:
encoded = _public_key(key).public_bytes(Encoding.OpenSSH, PublicFormat.OpenSSH).decode("ascii")
return f"{encoded} {comment}".strip()
def ssh_private(key: PrivateKey, password: Optional[str] = None) -> bytes:
"""OpenSSH private key format, via paramiko."""
import paramiko
key = _require_private(key)
if isinstance(key, rsa.RSAPrivateKey):
pkey = paramiko.RSAKey(key=key)
elif isinstance(key, dsa.DSAPrivateKey):
pkey = paramiko.DSSKey(key=key)
elif isinstance(key, ec.EllipticCurvePrivateKey):
pkey = paramiko.ECDSAKey(key=key)
elif isinstance(key, ed25519.Ed25519PrivateKey):
pkey = paramiko.Ed25519Key(key=key)
else: # pragma: no cover - defensive
raise ValueError("unsupported key type for SSH export")
buffer = io.StringIO()
pkey.write_private_key(buffer, password=password)
return buffer.getvalue().encode("utf-8")
# ---------------------------------------------------------------------------
# JSON Web Key (RFC 7517)
# ---------------------------------------------------------------------------
def _b64u(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
def _int_bytes(value: int) -> bytes:
return value.to_bytes((value.bit_length() + 7) // 8 or 1, "big")
def jwk_public(key: PrivateKey | PublicKey, *, kid: Optional[str] = None) -> Dict[str, Any]:
pub = _public_key(key)
if isinstance(pub, rsa.RSAPublicKey):
numbers = pub.public_numbers()
result: Dict[str, Any] = {
"kty": "RSA",
"n": _b64u(_int_bytes(numbers.n)),
"e": _b64u(_int_bytes(numbers.e)),
}
elif isinstance(pub, ec.EllipticCurvePublicKey):
numbers = pub.public_numbers()
curve = {
"secp256r1": "P-256",
"secp384r1": "P-384",
"secp521r1": "P-521",
"secp256k1": "secp256k1",
}.get(pub.curve.name, pub.curve.name)
result = {
"kty": "EC",
"crv": curve,
"x": _b64u(numbers.x.to_bytes((numbers.x.bit_length() + 7) // 8 or 1, "big")),
"y": _b64u(numbers.y.to_bytes((numbers.y.bit_length() + 7) // 8 or 1, "big")),
}
elif isinstance(pub, ed25519.Ed25519PublicKey):
result = {
"kty": "OKP",
"crv": "Ed25519",
"x": _b64u(pub.public_bytes(Encoding.Raw, PublicFormat.Raw)),
}
else:
raise ValueError("JWK is not supported for this key type")
if kid:
result["kid"] = kid
return result
def jwk_private(key: PrivateKey, *, kid: Optional[str] = None) -> Dict[str, Any]:
key = _require_private(key)
result = jwk_public(key, kid=kid)
if isinstance(key, rsa.RSAPrivateKey):
numbers = key.private_numbers()
result.update(
{
"d": _b64u(_int_bytes(numbers.d)),
"p": _b64u(_int_bytes(numbers.p)),
"q": _b64u(_int_bytes(numbers.q)),
"dp": _b64u(_int_bytes(numbers.dmp1)),
"dq": _b64u(_int_bytes(numbers.dmq1)),
"qi": _b64u(_int_bytes(numbers.iqmp)),
}
)
elif isinstance(key, ec.EllipticCurvePrivateKey):
result["d"] = _b64u(_int_bytes(key.private_numbers().private_value))
elif isinstance(key, ed25519.Ed25519PrivateKey):
result["d"] = _b64u(key.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption()))
else:
raise ValueError("JWK private export is not supported for this key type")
return result
# ---------------------------------------------------------------------------
# Microsoft PVK (RSA)
# ---------------------------------------------------------------------------
_PVK_MAGIC = 0xB0B5F11E
_PVK_KEYTYPE = 2
_CALG_RSA_KEYX = 0x00002400
_RSA2 = 0x32415352
def pvk_encode_private(key: PrivateKey) -> bytes:
if not isinstance(key, rsa.RSAPrivateKey):
raise ValueError("PVK export supports RSA keys only")
numbers = key.private_numbers()
pub = numbers.public_numbers
size = (pub.n.bit_length() + 7) // 8
blob = struct.pack("<BBBBL", 0x07, 0x02, 0, 0, _CALG_RSA_KEYX)
blob += struct.pack("<LLL", _RSA2, size * 8, pub.e)
for value in (
pub.n,
numbers.p,
numbers.q,
numbers.dmp1,
numbers.dmq1,
numbers.iqmp,
numbers.d,
):
blob += value.to_bytes(size, "little")
header = struct.pack("<LLLLLL", _PVK_MAGIC, 0, _PVK_KEYTYPE, 0, 0, len(blob))
return header + blob
def pvk_decode_private(data: bytes) -> rsa.RSAPrivateKey:
if len(data) < 24:
raise ValueError("not a PVK file")
magic, _reserved, key_type, encrypted, encrypted_len, blob_len = struct.unpack("<LLLLLL", data[:24])
if magic != _PVK_MAGIC or key_type != _PVK_KEYTYPE or encrypted:
raise ValueError("unsupported PVK file")
blob = data[24 : 24 + blob_len]
if len(blob) < 20:
raise ValueError("truncated PVK file")
_btype, _bversion, _r1, _r2, _alg = struct.unpack("<BBBBL", blob[:8])
magic2, bits, pubexp = struct.unpack("<LLL", blob[8:20])
if magic2 != _RSA2:
raise ValueError("unsupported PVK key blob")
size = (bits + 7) // 8
pos = 20
values = []
for _ in range(7):
chunk = blob[pos : pos + size]
if len(chunk) != size:
raise ValueError("truncated PVK key")
values.append(int.from_bytes(chunk, "little"))
pos += size
n, p, q, dmp1, dmq1, iqmp, d = values
return rsa.RSAPrivateNumbers(
p=p,
q=q,
d=d,
dmp1=dmp1,
dmq1=dmq1,
iqmp=iqmp,
public_numbers=rsa.RSAPublicNumbers(e=pubexp, n=n),
).private_key()
# ---------------------------------------------------------------------------
# Parsing
# ---------------------------------------------------------------------------
def load_key_bytes(data: bytes, password: Optional[str] = None) -> Tuple[PrivateKey | PublicKey, bool]:
"""Load a key from PEM/DER/OpenSSH/PVK bytes.
Returns ``(key, is_private)``. Raises :class:`ValueError` when nothing
matches.
"""
password_bytes = password.encode() if password else None
errors: List[str] = []
# PEM private / public
try:
key = serialization.load_pem_private_key(data, password=password_bytes)
return key, True
except Exception as exc: # noqa: BLE001 - probing
errors.append(str(exc))
try:
key = serialization.load_pem_public_key(data)
return key, False
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
# DER private / public
try:
key = serialization.load_der_private_key(data, password=password_bytes)
return key, True
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
try:
key = serialization.load_der_public_key(data)
return key, False
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
# OpenSSH private / public
try:
key = serialization.load_ssh_private_key(data, password=password_bytes)
return key, True
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
try:
key = serialization.load_ssh_public_key(data)
return key, False
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
# Microsoft PVK
try:
return pvk_decode_private(data), True
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
# Encrypted OpenSSH keys are not supported by cryptography; paramiko can
# read them. Convert the common key types back to cryptography objects.
try:
return _load_ssh_private_with_paramiko(data, password), True
except Exception as exc: # noqa: BLE001
errors.append(str(exc))
raise ValueError("unable to parse key: " + (errors[-1] if errors else "unknown format"))
def _load_ssh_private_with_paramiko(data: bytes, password: Optional[str]) -> PrivateKey:
import paramiko
text = data.decode("utf-8", "replace")
last: Optional[Exception] = None
for cls in (paramiko.RSAKey, paramiko.ECDSAKey, paramiko.Ed25519Key, paramiko.DSSKey):
try:
pkey = cls.from_private_key(io.StringIO(text), password=password)
break
except Exception as exc: # noqa: BLE001
last = exc
else: # pragma: no cover - defensive
raise ValueError(f"unable to parse OpenSSH private key: {last}")
# Paramiko keeps the cryptography signing key on its instances. This is
# stable across 3.x and avoids re-implementing key conversion.
signing_key = getattr(pkey, "_signing_key", None)
if signing_key is not None and is_private_key(signing_key):
return signing_key
if isinstance(pkey, paramiko.RSAKey):
pub = rsa.RSAPublicNumbers(e=pkey.e, n=pkey.n)
return rsa.RSAPrivateNumbers(
p=pkey.p, q=pkey.q, d=pkey.d, dmp1=pkey.dmp1, dmq1=pkey.dmq1, iqmp=pkey.iqmp,
public_numbers=pub,
).private_key()
if isinstance(pkey, paramiko.DSSKey):
pub = dsa.DSAPublicNumbers(parameter_numbers=dsa.DSAParameterNumbers(pkey.p, pkey.q, pkey.g), y=pkey.y)
return dsa.DSAPrivateNumbers(x=pkey.x, public_numbers=pub).private_key()
raise ValueError("unable to convert the SSH private key")
def public_key_from_der(data: bytes) -> PublicKey:
return serialization.load_der_public_key(data)
def private_key_from_der(data: bytes, password: Optional[str] = None) -> PrivateKey:
return serialization.load_der_private_key(data, password=password.encode() if password else None)
def public_key_sha256(key: PrivateKey | PublicKey) -> str:
return hashlib.sha256(public_der(key)).hexdigest()
def key_type_options() -> List[Dict[str, Any]]:
return [
{"type": "rsa", "label": "RSA", "sizes": [1024, 2048, 3072, 4096, 8192]},
{"type": "dsa", "label": "DSA", "sizes": [1024, 2048, 3072]},
{"type": "ec", "label": "EC", "curves": list(EC_CURVES)},
{"type": "ed25519", "label": "ED25519", "sizes": []},
]
__all__ = [
"DEFAULT_KEY_SIZE",
"EC_CURVES",
"HASH_ALGORITHMS",
"KEY_TYPES",
"KeyInfo",
"MAX_KEY_SIZE",
"MIN_KEY_SIZE",
"PrivateKey",
"PublicKey",
"curve_name",
"generate_key",
"hash_algorithm",
"is_private_key",
"jwk_private",
"jwk_public",
"key_info",
"key_size",
"key_type_name",
"key_type_options",
"load_key_bytes",
"possible_hashes",
"private_der",
"private_der_bytes",
"private_key_from_der",
"private_pem",
"public_der",
"public_der_bytes",
"public_key_from_der",
"public_key_sha256",
"public_pem",
"pvk_decode_private",
"pvk_encode_private",
"ssh_private",
"ssh_public",
]
@@ -0,0 +1,260 @@
"""Distinguished names and the small OID registry used by the UI.
XCA stores a subject as an ordered sequence of X.509 name entries and lets the
user add the same attribute more than once. ``cryptography`` models that as a
sequence of RDNs; we build one RDN per entry so order and duplicates survive a
round-trip.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
from cryptography import x509
from cryptography.x509.oid import NameOID, ObjectIdentifier
#: Common distinguished-name attributes. The value is ``(short name, long name)``.
DN_ATTRIBUTES: Dict[str, Tuple[str, str]] = {
"2.5.4.3": ("CN", "commonName"),
"2.5.4.4": ("SN", "surname"),
"2.5.4.5": ("serialNumber", "serialNumber"),
"2.5.4.6": ("C", "countryName"),
"2.5.4.7": ("L", "localityName"),
"2.5.4.8": ("ST", "stateOrProvinceName"),
"2.5.4.9": ("street", "streetAddress"),
"2.5.4.10": ("O", "organizationName"),
"2.5.4.11": ("OU", "organizationalUnitName"),
"2.5.4.12": ("title", "title"),
"2.5.4.17": ("postalCode", "postalCode"),
"2.5.4.42": ("GN", "givenName"),
"2.5.4.43": ("initials", "initials"),
"2.5.4.44": ("generationQualifier", "generationQualifier"),
"2.5.4.46": ("dnQualifier", "dnQualifier"),
"2.5.4.65": ("pseudonym", "pseudonym"),
"2.5.4.97": ("organizationIdentifier", "organizationIdentifier"),
"1.2.840.113549.1.9.1": ("emailAddress", "emailAddress"),
"0.9.2342.19200300.100.1.1": ("UID", "userId"),
"0.9.2342.19200300.100.1.25": ("DC", "domainComponent"),
}
#: Aliases accepted from the API / template format.
DN_ALIASES: Dict[str, str] = {
"cn": "2.5.4.3",
"commonname": "2.5.4.3",
"surname": "2.5.4.4",
"sn": "2.5.4.4",
"serialnumber": "2.5.4.5",
"c": "2.5.4.6",
"countryname": "2.5.4.6",
"l": "2.5.4.7",
"localityname": "2.5.4.7",
"st": "2.5.4.8",
"stateorprovincename": "2.5.4.8",
"street": "2.5.4.9",
"streetaddress": "2.5.4.9",
"o": "2.5.4.10",
"organizationname": "2.5.4.10",
"ou": "2.5.4.11",
"organizationalunitname": "2.5.4.11",
"title": "2.5.4.12",
"postalcode": "2.5.4.17",
"gn": "2.5.4.42",
"givenname": "2.5.4.42",
"initials": "2.5.4.43",
"generationqualifier": "2.5.4.44",
"dnqualifier": "2.5.4.46",
"pseudonym": "2.5.4.65",
"organizationidentifier": "2.5.4.97",
"email": "1.2.840.113549.1.9.1",
"emailaddress": "1.2.840.113549.1.9.1",
"uid": "0.9.2342.19200300.100.1.1",
"userid": "0.9.2342.19200300.100.1.1",
"dc": "0.9.2342.19200300.100.1.25",
"domaincomponent": "0.9.2342.19200300.100.1.25",
}
#: Extended-key-usage names accepted in specs and templates.
EKU_OIDS: Dict[str, str] = {
"serverAuth": "1.3.6.1.5.5.7.3.1",
"clientAuth": "1.3.6.1.5.5.7.3.2",
"codeSigning": "1.3.6.1.5.5.7.3.3",
"emailProtection": "1.3.6.1.5.5.7.3.4",
"ipsecEndSystem": "1.3.6.1.5.5.7.3.5",
"ipsecTunnel": "1.3.6.1.5.5.7.3.6",
"ipsecUser": "1.3.6.1.5.5.7.3.7",
"timeStamping": "1.3.6.1.5.5.7.3.8",
"OCSPSigning": "1.3.6.1.5.5.7.3.9",
"id-kp-eapOverPPP": "1.3.6.1.5.5.7.3.13",
"id-kp-eapOverLAN": "1.3.6.1.5.5.7.3.14",
"secureShellClient": "1.3.6.1.5.5.7.3.21",
"secureShellServer": "1.3.6.1.5.5.7.3.22",
"msCodeInd": "1.3.6.1.4.1.311.2.1.21",
"msCodeCom": "1.3.6.1.4.1.311.2.1.22",
"msCTLSign": "1.3.6.1.4.1.311.10.3.1",
"msSGC": "1.3.6.1.4.1.311.10.3.3",
"msEFS": "1.3.6.1.4.1.311.10.3.4",
"nsSGC": "2.16.840.1.113730.4.1",
"pkInitKDC": "1.3.6.1.5.2.3.5",
"adobePDFSigning": "1.2.840.113583.1.1.5",
"msofficeSigning": "1.3.6.1.4.1.311.10.3.12",
}
_EKU_BY_OID = {oid: name for name, oid in EKU_OIDS.items()}
@dataclass(frozen=True)
class DnEntry:
oid: str
name: str
value: str
def as_dict(self) -> Dict[str, str]:
return {"oid": self.oid, "name": self.name, "value": self.value}
def resolve_oid(value: str) -> str:
"""Resolve a short name, long name or dotted OID to a dotted OID."""
value = (value or "").strip()
if not value:
raise ValueError("empty attribute name")
if value[0].isdigit() and value.count(".") >= 1:
# Validate through cryptography so malformed OIDs fail early.
ObjectIdentifier(value)
return value
key = value.lower().replace(" ", "").replace("-", "").replace("_", "")
if key in DN_ALIASES:
return DN_ALIASES[key]
if value in DN_ATTRIBUTES:
return value
raise ValueError(f"unknown distinguished-name attribute {value!r}")
def oid_name(oid: str) -> str:
"""The short display name for an attribute OID."""
return DN_ATTRIBUTES.get(oid, (oid, oid))[0]
def oid_long_name(oid: str) -> str:
return DN_ATTRIBUTES.get(oid, (oid, oid))[1]
def entry_from_dict(entry: Mapping[str, Any]) -> DnEntry:
oid = resolve_oid(str(entry.get("oid") or entry.get("name") or ""))
value = entry.get("value", "")
if not isinstance(value, str):
value = str(value)
return DnEntry(oid=oid, name=oid_name(oid), value=value.strip())
def entries_from_json(entries: Iterable[Mapping[str, Any]]) -> List[DnEntry]:
return [entry_from_dict(entry) for entry in entries]
def name_from_entries(entries: Sequence[DnEntry]) -> x509.Name:
"""Build a Name, one RDN per entry, preserving order and duplicates."""
rdns = []
for entry in entries:
if not entry.value:
continue
rdns.append(
x509.RelativeDistinguishedName(
[x509.NameAttribute(ObjectIdentifier(entry.oid), entry.value)]
)
)
return x509.Name(rdns)
def entries_from_name(name: x509.Name) -> List[DnEntry]:
"""Flatten a Name into ordered entries (one per attribute)."""
result: List[DnEntry] = []
for rdn in name.rdns:
for attribute in rdn:
oid = attribute.oid.dotted_string
result.append(DnEntry(oid=oid, name=oid_name(oid), value=str(attribute.value)))
return result
def name_from_der(data: bytes) -> x509.Name:
return x509.Name.from_bytes(data)
def name_to_der(name: x509.Name) -> bytes:
return name.public_bytes()
def rfc4514(name: x509.Name) -> str:
return name.rfc4514_string()
def one_line(name: x509.Name) -> str:
"""A stable, human-readable one-line rendering (RFC 4514 with commas)."""
return name.rfc4514_string()
def name_hash(name: x509.Name) -> str:
"""SHA-256 of the DER encoding, used for fast issuer lookup."""
import hashlib
return hashlib.sha256(name.public_bytes()).hexdigest()
def common_names(name: x509.Name) -> List[str]:
return [str(a.value) for a in name.get_attributes_for_oid(NameOID.COMMON_NAME)]
def eku_name(oid: str) -> str:
return _EKU_BY_OID.get(oid, oid)
def eku_oid(name: str) -> str:
if name in EKU_OIDS:
return EKU_OIDS[name]
return resolve_oid(name)
def name_attribute_options() -> List[Dict[str, str]]:
"""The DN attributes offered by the UI, in the usual XCA order."""
order = [
"2.5.4.6",
"2.5.4.8",
"2.5.4.7",
"2.5.4.10",
"2.5.4.11",
"2.5.4.3",
"1.2.840.113549.1.9.1",
]
seen = set(order)
ordered = order + [oid for oid in DN_ATTRIBUTES if oid not in seen]
return [
{"oid": oid, "name": DN_ATTRIBUTES[oid][0], "long_name": DN_ATTRIBUTES[oid][1]}
for oid in ordered
]
def eku_options() -> List[Dict[str, str]]:
return [{"name": name, "oid": oid} for name, oid in EKU_OIDS.items()]
__all__ = [
"DN_ALIASES",
"DN_ATTRIBUTES",
"EKU_OIDS",
"DnEntry",
"common_names",
"eku_name",
"eku_oid",
"eku_options",
"entries_from_json",
"entries_from_name",
"entry_from_dict",
"name_attribute_options",
"name_from_der",
"name_from_entries",
"name_hash",
"name_to_der",
"oid_long_name",
"oid_name",
"one_line",
"resolve_oid",
"rfc4514",
]
@@ -0,0 +1,224 @@
"""PKCS#10 certificate signing request helpers."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Mapping, Optional
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from cryptography.x509.oid import ObjectIdentifier
from .extensions import build_extensions, extension_config_lines, extension_rows, extension_specs
from .keys import PrivateKey, PublicKey, hash_algorithm, public_der
from .names import entries_from_name, name_from_entries
CHALLENGE_PASSWORD_OID = "1.2.840.113549.1.9.7"
UNSTRUCTURED_NAME_OID = "1.2.840.113549.1.9.2"
ATTRIBUTE_NAMES = {
CHALLENGE_PASSWORD_OID: "Challenge Password",
UNSTRUCTURED_NAME_OID: "Unstructured Name",
}
@dataclass(frozen=True)
class RequestInfo:
subject: x509.Name
signature_algorithm: str
public_key_der: bytes
fingerprint_sha256: str
attributes: List[Dict[str, Any]]
def build_csr(
*,
subject: x509.Name,
private_key: PrivateKey,
extension_spec: Optional[Mapping[str, Any]] = None,
signature_hash: Optional[str] = "sha256",
attributes: Optional[Mapping[str, str]] = None,
) -> x509.CertificateSigningRequest:
public_key = private_key.public_key()
extensions = build_extensions(
extension_spec,
public_key=public_key,
subject_name=subject,
include_aki=False,
)
# A CSR must not carry an Authority Key Identifier or an issuer alt name;
# XCA removes them as well.
extensions = [
ext
for ext in extensions
if ext.oid
not in (
x509.oid.ExtensionOID.AUTHORITY_KEY_IDENTIFIER,
x509.oid.ExtensionOID.ISSUER_ALTERNATIVE_NAME,
)
]
builder = x509.CertificateSigningRequestBuilder().subject_name(subject)
for extension in extensions:
builder = builder.add_extension(extension.value, extension.critical)
for oid, value in (attributes or {}).items():
if value:
builder = builder.add_attribute(ObjectIdentifier(oid), str(value).encode("utf-8"))
algorithm = hash_algorithm(signature_hash, private_key)
return builder.sign(private_key, algorithm)
def parse_csr(data: bytes) -> x509.CertificateSigningRequest:
return x509.load_der_x509_csr(data)
def csr_der(csr: x509.CertificateSigningRequest) -> bytes:
return csr.public_bytes(serialization.Encoding.DER)
def csr_pem(csr: x509.CertificateSigningRequest) -> str:
return csr.public_bytes(serialization.Encoding.PEM).decode("ascii")
def verify_csr(csr: x509.CertificateSigningRequest) -> bool:
return bool(csr.is_signature_valid)
def request_info(csr: x509.CertificateSigningRequest) -> RequestInfo:
public_key = csr.public_key()
return RequestInfo(
subject=csr.subject,
signature_algorithm=_signature_name(csr),
public_key_der=public_der(public_key),
fingerprint_sha256=hashlib.sha256(csr_der(csr)).hexdigest(),
attributes=attribute_rows(csr),
)
_SIGNATURE_OIDS = {
"1.2.840.113549.1.1.5": "sha1WithRSAEncryption",
"1.2.840.113549.1.1.11": "sha256WithRSAEncryption",
"1.2.840.113549.1.1.12": "sha384WithRSAEncryption",
"1.2.840.113549.1.1.13": "sha512WithRSAEncryption",
"1.2.840.113549.1.1.14": "sha224WithRSAEncryption",
"1.2.840.10045.4.1": "ecdsa-with-SHA1",
"1.2.840.10045.4.3.2": "ecdsa-with-SHA256",
"1.2.840.10045.4.3.3": "ecdsa-with-SHA384",
"1.2.840.10045.4.3.4": "ecdsa-with-SHA512",
"1.3.101.112": "Ed25519",
}
def _signature_name(csr: x509.CertificateSigningRequest) -> str:
oid = csr.signature_algorithm_oid
return _SIGNATURE_OIDS.get(oid.dotted_string, getattr(oid, "name", None) or oid.dotted_string)
def _decode_attribute_value(value: bytes) -> str:
# cryptography returns the DER-encoded value bytes. Try UTF-8 first,
# then strip a simple ASN.1 string tag/length header.
try:
return value.decode("utf-8")
except UnicodeDecodeError:
pass
if len(value) >= 2 and value[0] in (0x0C, 0x13, 0x16, 0x1E):
length = value[1]
offset = 2
if length & 0x80:
count = length & 0x7F
length = int.from_bytes(value[2 : 2 + count], "big")
offset = 2 + count
raw = value[offset : offset + length]
if value[0] == 0x1E:
return raw.decode("utf-16-be", "replace")
return raw.decode("utf-8", "replace")
return value.hex()
def attribute_rows(csr: x509.CertificateSigningRequest) -> List[Dict[str, Any]]:
rows = []
for attribute in csr.attributes:
oid = attribute.oid.dotted_string
rows.append(
{
"oid": oid,
"name": ATTRIBUTE_NAMES.get(oid, oid),
"values": [_decode_attribute_value(attribute.value)],
}
)
return rows
def attribute_values(csr: x509.CertificateSigningRequest) -> Dict[str, str]:
result = {}
for row in attribute_rows(csr):
if row["values"]:
result[row["oid"]] = row["values"][0]
return result
def csr_spec(csr: x509.CertificateSigningRequest) -> Dict[str, Any]:
return {
"subject": [entry.as_dict() for entry in entries_from_name(csr.subject)],
"signature_hash": _hash_name(csr),
"extensions": extension_specs(csr.extensions),
"attributes": attribute_values(csr),
"public_key_der": public_der(csr.public_key()).hex(),
}
def _hash_name(csr: x509.CertificateSigningRequest) -> Optional[str]:
try:
algorithm = csr.signature_hash_algorithm
except Exception: # noqa: BLE001 - Ed25519
return None
return algorithm.name if algorithm is not None else None
def csr_extensions(csr: x509.CertificateSigningRequest) -> List[Dict[str, Any]]:
return extension_rows(csr.extensions)
def csr_extension_specs(csr: x509.CertificateSigningRequest) -> Dict[str, Any]:
return extension_specs(csr.extensions)
def csr_public_key(csr: x509.CertificateSigningRequest) -> PublicKey:
return csr.public_key()
def openssl_config(csr: x509.CertificateSigningRequest) -> str:
lines = [
"[req]",
"distinguished_name = req_distinguished_name",
"prompt = no",
"",
"[req_distinguished_name]",
]
for entry in entries_from_name(csr.subject):
lines.append(f"{entry.name} = {entry.value}")
lines.extend(["", "[v3_ext]"])
config = extension_config_lines(csr.extensions)
if config:
lines.append(config)
return "\n".join(lines) + "\n"
__all__ = [
"ATTRIBUTE_NAMES",
"CHALLENGE_PASSWORD_OID",
"RequestInfo",
"UNSTRUCTURED_NAME_OID",
"attribute_rows",
"attribute_values",
"build_csr",
"csr_der",
"csr_extension_specs",
"csr_extensions",
"csr_pem",
"csr_public_key",
"csr_spec",
"openssl_config",
"parse_csr",
"request_info",
"verify_csr",
]
@@ -0,0 +1,583 @@
"""XCA-style certificate/request templates.
A template is a subject plus a set of extension defaults and a relative
validity range. Built-in templates mirror XCA's ``CA``, ``TLS_server`` and
``TLS_client`` templates; user templates are stored as JSON by the app.
"""
from __future__ import annotations
import base64
import binascii
import struct
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence
from cryptography import x509
from .extensions import (
default_extension_spec,
parse_aia_descriptions,
parse_crl_distribution_urls,
parse_eku_names,
parse_general_name_text,
parse_key_usage_names,
parse_name_constraints_text,
)
from .names import DnEntry, entries_from_json, entries_from_name, name_from_entries
TEMPLATE_VERSION = 10
PEM_STRING_XCA_TEMPLATE = "XCA TEMPLATE"
@dataclass(frozen=True)
class Template:
id: str
name: str
comment: str
subject: List[Dict[str, str]]
settings: Dict[str, Any]
builtin: bool = False
def as_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"comment": self.comment,
"subject": self.subject,
"settings": self.settings,
"builtin": self.builtin,
}
def default_settings() -> Dict[str, Any]:
extensions = default_extension_spec()
extensions["basic_constraints"] = {"critical": True, "ca": None, "path_length": None}
return {
"extensions": extensions,
"validity": {"amount": 365, "unit": "days", "midnight": False, "no_expiration": False},
}
def _base_settings() -> Dict[str, Any]:
return default_settings()
def builtin_templates() -> List[Template]:
ca = _base_settings()
ca["extensions"].update(
{
"basic_constraints": {"critical": True, "ca": True, "path_length": None},
"key_usage": {"critical": True, "usages": ["key_cert_sign", "crl_sign"]},
"extended_key_usage": {"critical": False, "usages": []},
"subject_key_identifier": {"critical": False},
"authority_key_identifier": {"critical": False},
}
)
ca["validity"] = {"amount": 10, "unit": "years", "midnight": False, "no_expiration": False}
server = _base_settings()
server["extensions"].update(
{
"basic_constraints": {"critical": True, "ca": False, "path_length": None},
"key_usage": {
"critical": True,
"usages": ["digital_signature", "content_commitment", "key_encipherment", "key_agreement"],
},
"extended_key_usage": {"critical": False, "usages": ["serverAuth"]},
"subject_alt_name": {"critical": False, "entries": [{"type": "dns", "value": "copycn"}]},
"subject_key_identifier": {"critical": False},
"authority_key_identifier": {"critical": False},
}
)
server["validity"] = {"amount": 365, "unit": "days", "midnight": False, "no_expiration": False}
client = _base_settings()
client["extensions"].update(
{
"basic_constraints": {"critical": True, "ca": False, "path_length": None},
"key_usage": {
"critical": True,
"usages": ["digital_signature", "key_encipherment", "data_encipherment", "key_agreement"],
},
"extended_key_usage": {"critical": False, "usages": ["clientAuth"]},
"subject_key_identifier": {"critical": False},
"authority_key_identifier": {"critical": False},
}
)
client["validity"] = {"amount": 365, "unit": "days", "midnight": False, "no_expiration": False}
return [
Template(
id="builtin:ca",
name="CA",
comment="Default XCA CA template",
subject=[],
settings=ca,
builtin=True,
),
Template(
id="builtin:tls-server",
name="TLS server",
comment="Default XCA TLS server template",
subject=[],
settings=server,
builtin=True,
),
Template(
id="builtin:tls-client",
name="TLS client",
comment="Default XCA TLS client template",
subject=[],
settings=client,
builtin=True,
),
]
def apply_template(
template: Mapping[str, Any],
target: Mapping[str, Any],
*,
apply_subject: bool = True,
apply_extensions: bool = True,
) -> Dict[str, Any]:
"""Merge a template into a certificate/request creation spec."""
result = dict(target)
if apply_subject and template.get("subject"):
result["subject"] = [dict(item) for item in template["subject"]]
if apply_extensions:
settings = template.get("settings") or {}
extensions = settings.get("extensions") or {}
merged = dict(result.get("extensions") or {})
for key, value in extensions.items():
if value is not None:
merged[key] = value
result["extensions"] = merged
if settings.get("validity"):
result["validity"] = dict(settings["validity"])
return result
def template_from_certificate(cert: x509.Certificate, name: str, comment: str = "") -> Dict[str, Any]:
from .certificates import certificate_extension_specs
return {
"name": name,
"comment": comment,
"subject": [entry.as_dict() for entry in entries_from_name(cert.subject)],
"settings": {
"extensions": certificate_extension_specs(cert),
"validity": {"amount": 365, "unit": "days", "midnight": False, "no_expiration": False},
},
}
def template_from_request(csr: x509.CertificateSigningRequest, name: str, comment: str = "") -> Dict[str, Any]:
from .requests import csr_extension_specs
return {
"name": name,
"comment": comment,
"subject": [entry.as_dict() for entry in entries_from_name(csr.subject)],
"settings": {
"extensions": csr_extension_specs(csr),
"validity": {"amount": 365, "unit": "days", "midnight": False, "no_expiration": False},
},
}
# ---------------------------------------------------------------------------
# XCA template binary format (Qt QDataStream, version 10)
# ---------------------------------------------------------------------------
def _qstring(value: Optional[str]) -> bytes:
if value is None:
return struct.pack(">I", 0xFFFFFFFF)
raw = value.encode("utf-16-be")
return struct.pack(">I", len(raw)) + raw
def _read_qstring(data: bytes, offset: int) -> tuple[Optional[str], int]:
if offset + 4 > len(data):
raise ValueError("truncated template")
length = struct.unpack(">I", data[offset : offset + 4])[0]
offset += 4
if length == 0xFFFFFFFF:
return None, offset
if offset + length > len(data):
raise ValueError("truncated template string")
raw = data[offset : offset + length]
offset += length
return raw.decode("utf-16-be", "replace"), offset
def _qmap(values: Mapping[str, str]) -> bytes:
# QMap iterates in key order.
keys = sorted(values)
out = struct.pack(">I", len(keys))
for key in keys:
out += _qstring(key) + _qstring(values[key])
return out
def _read_qmap(data: bytes, offset: int) -> tuple[Dict[str, str], int]:
if offset + 4 > len(data):
raise ValueError("truncated template map")
count = struct.unpack(">I", data[offset : offset + 4])[0]
offset += 4
result: Dict[str, str] = {}
for _ in range(count):
key, offset = _read_qstring(data, offset)
value, offset = _read_qstring(data, offset)
if key is not None:
result[key] = value or ""
return result, offset
def _der_name_length(data: bytes) -> int:
if len(data) < 2 or data[0] != 0x30:
raise ValueError("template does not start with an X.509 name")
first = data[1]
if first < 0x80:
return 2 + first
count = first & 0x7F
if count == 0 or len(data) < 2 + count:
raise ValueError("invalid DER length")
length = int.from_bytes(data[2 : 2 + count], "big")
return 2 + count + length
def template_to_xca_data(template: Mapping[str, Any]) -> bytes:
"""Serialize a template in XCA's PEM-like template format."""
subject = name_from_entries(entries_from_json(template.get("subject") or []))
name_der = subject.public_bytes()
settings = template_to_xca_settings(template)
settings["internal_name"] = str(template.get("name") or "")
settings["internal_comment"] = str(template.get("comment") or "")
data = name_der + _qmap(settings)
payload = struct.pack(">I", len(data)) + struct.pack(">I", TEMPLATE_VERSION) + data
body = base64.encodebytes(payload).decode("ascii").replace("\n", "")
wrapped = "\n".join(body[i : i + 64] for i in range(0, len(body), 64))
return (
f"-----BEGIN {PEM_STRING_XCA_TEMPLATE}-----\n"
f"{wrapped}\n"
f"-----END {PEM_STRING_XCA_TEMPLATE}-----\n"
).encode("ascii")
def template_from_xca_data(data: bytes | str) -> Dict[str, Any]:
"""Parse XCA's PEM-like template format (version 10)."""
if isinstance(data, bytes):
text = data.decode("ascii", "replace")
else:
text = data
begin = f"-----BEGIN {PEM_STRING_XCA_TEMPLATE}-----"
end = f"-----END {PEM_STRING_XCA_TEMPLATE}-----"
if begin not in text or end not in text:
raise ValueError("not an XCA template")
body = text.split(begin, 1)[1].split(end, 1)[0]
try:
payload = base64.b64decode("".join(body.split()))
except binascii.Error as exc:
raise ValueError("invalid base64 in XCA template") from exc
if len(payload) < 8:
raise ValueError("truncated XCA template")
_size, version = struct.unpack(">II", payload[:8])
if version != TEMPLATE_VERSION:
raise ValueError(f"unsupported XCA template version {version}")
content = payload[8:]
name_len = _der_name_length(content)
name = x509.Name.from_bytes(content[:name_len])
settings, _offset = _read_qmap(content, name_len)
name_value = settings.pop("internal_name", "")
comment = settings.pop("internal_comment", "")
settings_json = xca_settings_to_template(settings)
return {
"name": name_value or "XCA template",
"comment": comment,
"subject": [entry.as_dict() for entry in entries_from_name(name)],
"settings": settings_json,
}
def template_from_xca_payload(data: bytes, version: int = TEMPLATE_VERSION) -> Dict[str, Any]:
"""Parse the raw ``templates.template`` payload from an XCA database."""
if version != TEMPLATE_VERSION:
raise ValueError(f"unsupported XCA template version {version}")
name_len = _der_name_length(data)
name = x509.Name.from_bytes(data[:name_len])
settings, _offset = _read_qmap(data, name_len)
name_value = settings.pop("internal_name", "")
comment = settings.pop("internal_comment", "")
settings_json = xca_settings_to_template(settings)
return {
"name": name_value or "XCA template",
"comment": comment,
"subject": [entry.as_dict() for entry in entries_from_name(name)],
"settings": settings_json,
}
def xca_settings_to_template(settings: Mapping[str, str]) -> Dict[str, Any]:
"""Convert XCA's ``pki_temp`` settings to the KCA template JSON."""
result = default_settings()
extensions = result["extensions"]
extensions["basic_constraints"] = None
ca = settings.get("ca", "")
if ca == "1":
extensions["basic_constraints"] = {
"critical": settings.get("bcCritical") == "1",
"ca": True,
"path_length": int(settings["basicPath"]) if settings.get("basicPath", "").isdigit() else None,
}
elif ca == "2":
extensions["basic_constraints"] = {
"critical": settings.get("bcCritical") == "1",
"ca": False,
"path_length": None,
}
key_use = int(settings.get("keyUse") or 0)
usage_names = [
name
for index, name in enumerate(
[
"digital_signature",
"content_commitment",
"key_encipherment",
"data_encipherment",
"key_agreement",
"key_cert_sign",
"crl_sign",
"encipher_only",
"decipher_only",
]
)
if key_use & (1 << index)
]
extensions["key_usage"] = {
"critical": settings.get("kuCritical") == "1",
"usages": usage_names,
}
extensions["extended_key_usage"] = {
"critical": settings.get("ekuCritical") == "1",
"usages": parse_eku_names(settings.get("eKeyUse", "")),
}
if settings.get("subAltName"):
try:
extensions["subject_alt_name"] = {
"critical": False,
"entries": [
{"type": "dns", "value": "copycn"}
]
if settings.get("subAltName") == "DNS:copycn"
else [
{"type": _kind, "value": _value}
for _kind, _value in _parse_general_text(settings["subAltName"])
],
}
except ValueError:
extensions["subject_alt_name"] = {"critical": False, "entries": []}
if settings.get("issAltName"):
try:
extensions["issuer_alt_name"] = {
"critical": False,
"entries": [
{"type": _kind, "value": _value}
for _kind, _value in _parse_general_text(settings["issAltName"])
],
}
except ValueError:
pass
if settings.get("crlDist"):
extensions["crl_distribution_points"] = {
"critical": False,
"urls": parse_crl_distribution_urls(settings["crlDist"]),
}
if settings.get("authInfAcc"):
extensions["authority_information_access"] = {
"critical": False,
"descriptions": parse_aia_descriptions(settings["authInfAcc"]),
}
if settings.get("nameCons"):
extensions["name_constraints"] = {
"critical": True,
**parse_name_constraints_text(settings["nameCons"]),
}
extensions["subject_key_identifier"] = {"critical": False} if settings.get("subKey") == "1" else None
extensions["authority_key_identifier"] = {"critical": False} if settings.get("authKey") == "1" else None
if settings.get("OCSPstaple") == "1":
extensions["tls_feature"] = {"critical": False, "features": ["status_request"]}
ns_type = int(settings.get("nsCertType") or 0)
ns_names = [
name
for index, name in enumerate(["client", "server", "email", "objsign", "sslCA", "emailCA", "objCA"])
if ns_type & (1 << index)
]
netscape: Dict[str, Any] = {}
if ns_names:
netscape["cert_type"] = ns_names
ns_fields = {
"nsComment": "comment",
"nsBaseUrl": "base_url",
"nsRevocationUrl": "revocation_url",
"nsCARevocationUrl": "ca_revocation_url",
"nsRenewalUrl": "renewal_url",
"nsCaPolicyUrl": "ca_policy_url",
"nsSslServerName": "ssl_server_name",
}
for source, target in ns_fields.items():
if settings.get(source):
netscape[target] = settings[source]
if netscape:
extensions["netscape"] = netscape
if settings.get("adv_ext"):
# Kept for display/export; the typed editor does not interpret the
# OpenSSL config text.
result["advanced_config"] = settings["adv_ext"]
valid_n = int(settings.get("validN") or 365)
valid_m = int(settings.get("validM") or 0)
result["validity"] = {
"amount": valid_n,
"unit": "days" if valid_m == 0 else ("months" if valid_m == 1 else "years"),
"midnight": settings.get("validMidn") == "1",
"no_expiration": settings.get("noWellDefinedExpDate") == "1",
}
return result
def _parse_general_text(text: str) -> List[tuple[str, str]]:
result = []
for chunk in str(text).split(","):
chunk = chunk.strip()
if not chunk:
continue
if ":" not in chunk:
continue
prefix, value = chunk.split(":", 1)
prefix = prefix.strip().lower()
kind = {
"dns": "dns",
"ip": "ip",
"uri": "uri",
"email": "email",
"rid": "rid",
"othername": "other_name",
"dirname": "directory_name",
}.get(prefix)
if kind:
result.append((kind, value.strip()))
return result
def template_to_xca_settings(template: Mapping[str, Any]) -> Dict[str, str]:
"""Convert the KCA template JSON to XCA's ``pki_temp`` settings map."""
settings = dict(template.get("settings") or {})
extensions = settings.get("extensions") or {}
result: Dict[str, str] = {}
bc = extensions.get("basic_constraints") or {}
if bc.get("ca") is True:
result["ca"] = "1"
result["bcCritical"] = "1" if bc.get("critical") else "0"
result["basicPath"] = str(bc.get("path_length") or "")
elif bc.get("ca") is False:
result["ca"] = "2"
result["bcCritical"] = "1" if bc.get("critical") else "0"
result["basicPath"] = ""
else:
result["ca"] = ""
result["bcCritical"] = ""
result["basicPath"] = ""
usage_index = {
"digital_signature": 0,
"content_commitment": 1,
"key_encipherment": 2,
"data_encipherment": 3,
"key_agreement": 4,
"key_cert_sign": 5,
"crl_sign": 6,
"encipher_only": 7,
"decipher_only": 8,
}
ku = extensions.get("key_usage") or {}
result["keyUse"] = str(sum(1 << usage_index[name] for name in ku.get("usages", []) if name in usage_index))
result["kuCritical"] = "1" if ku.get("critical") else "0"
eku = extensions.get("extended_key_usage") or {}
result["eKeyUse"] = ", ".join(eku.get("usages", []))
result["ekuCritical"] = "1" if eku.get("critical") else "0"
result["subKey"] = "1" if extensions.get("subject_key_identifier") else "0"
result["authKey"] = "1" if extensions.get("authority_key_identifier") else "0"
san = extensions.get("subject_alt_name") or {}
result["subAltName"] = _general_text(san.get("entries", []))
ian = extensions.get("issuer_alt_name") or {}
result["issAltName"] = _general_text(ian.get("entries", []))
result["crlDist"] = ", ".join(
f"URI:{url}" for url in (extensions.get("crl_distribution_points") or {}).get("urls", [])
)
result["authInfAcc"] = ", ".join(
f"{'OCSP' if item.get('method') == 'ocsp' else 'caIssuers'};URI:{(item.get('location') or {}).get('value', '')}"
for item in (extensions.get("authority_information_access") or {}).get("descriptions", [])
)
nc = extensions.get("name_constraints") or {}
result["nameCons"] = ", ".join(
[f"permitted;{_one_general(item)}" for item in nc.get("permitted", [])]
+ [f"excluded;{_one_general(item)}" for item in nc.get("excluded", [])]
)
result["OCSPstaple"] = "1" if extensions.get("tls_feature") else "0"
ns = extensions.get("netscape") or {}
ns_index = {name: index for index, name in enumerate(["client", "server", "email", "objsign", "sslCA", "emailCA", "objCA"])}
result["nsCertType"] = str(sum(1 << ns_index[name] for name in ns.get("cert_type", []) if name in ns_index))
result["nsComment"] = str(ns.get("comment", ""))
result["nsBaseUrl"] = str(ns.get("base_url", ""))
result["nsRevocationUrl"] = str(ns.get("revocation_url", ""))
result["nsCARevocationUrl"] = str(ns.get("ca_revocation_url", ""))
result["nsRenewalUrl"] = str(ns.get("renewal_url", ""))
result["nsCaPolicyUrl"] = str(ns.get("ca_policy_url", ""))
result["nsSslServerName"] = str(ns.get("ssl_server_name", ""))
result["adv_ext"] = str(settings.get("advanced_config", ""))
validity = settings.get("validity") or {}
result["validN"] = str(validity.get("amount", 365))
result["validM"] = {"days": "0", "months": "1", "years": "2"}.get(str(validity.get("unit", "days")), "0")
result["validMidn"] = "1" if validity.get("midnight") else "0"
result["noWellDefinedExpDate"] = "1" if validity.get("no_expiration") else "0"
return result
def _general_text(entries: Sequence[Mapping[str, Any]]) -> str:
return ", ".join(_one_general(entry) for entry in entries)
def _one_general(entry: Mapping[str, Any]) -> str:
kind = str(entry.get("type", "dns"))
value = entry.get("value", "")
prefix = {
"dns": "DNS",
"ip": "IP",
"uri": "URI",
"email": "email",
"rid": "RID",
"other_name": "otherName",
"directory_name": "dirName",
}.get(kind, kind)
if kind == "other_name":
return f"{prefix}:{entry.get('oid', '')};UTF8:{value}"
return f"{prefix}:{value}"
__all__ = [
"PEM_STRING_XCA_TEMPLATE",
"TEMPLATE_VERSION",
"Template",
"apply_template",
"builtin_templates",
"default_settings",
"template_from_certificate",
"template_from_request",
"template_from_xca_data",
"template_from_xca_payload",
"template_to_xca_data",
"template_to_xca_settings",
"xca_settings_to_template",
]
@@ -0,0 +1,334 @@
"""Import an XCA SQLite database (``.xdb``).
XCA stores every cryptographic object as base64 DER in a small relational
schema. This reader supports schema 7/8 (XCA 2.x), software keys and the
normal item types; PKCS#11 token rows are skipped because a browser app cannot
use a card reader. Private keys are decrypted with the database password
(individual key passwords are tried with the same value and otherwise the
public part is imported).
"""
from __future__ import annotations
import base64
import hashlib
import sqlite3
import tempfile
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.ciphers import Cipher, modes
from cryptography.hazmat.decrepit.ciphers.algorithms import TripleDES
from .certificates import parse_certificate
from .crls import parse_crl
from .imports import ParsedItem
from .keys import is_private_key, load_key_bytes
from .requests import parse_csr
from .templates import template_from_xca_payload
_PKI_SOURCE = {
0: "unknown",
1: "imported",
2: "generated",
3: "transformed",
4: "token",
5: "legacy",
6: "renewed",
}
def _parse_date(value: Optional[str]) -> Optional[datetime]:
if not value:
return None
text = str(value).strip()
if not text:
return None
for fmt in ("%Y%m%d%H%M%SZ", "%Y%m%d%H%M%S"):
try:
return datetime.strptime(text, fmt).replace(tzinfo=timezone.utc)
except ValueError:
continue
return None
def _evp_bytes_to_key(password: bytes, salt: bytes, key_len: int = 24) -> bytes:
"""OpenSSL's EVP_BytesToKey with SHA-1 and one iteration."""
result = b""
previous = b""
while len(result) < key_len:
digest = hashlib.sha1()
digest.update(previous)
digest.update(password)
digest.update(salt)
previous = digest.digest()
result += previous
return result[:key_len]
def _legacy_decrypt(data: bytes, password: Optional[str]) -> Optional[bytes]:
if not password or len(data) <= 8:
return None
iv = data[:8]
ciphertext = data[8:]
key = _evp_bytes_to_key(password.encode(), iv, 24)
try:
decryptor = Cipher(TripleDES(key), modes.CBC(iv)).decryptor()
padded = decryptor.update(ciphertext) + decryptor.finalize()
pad = padded[-1]
if pad < 1 or pad > 8:
return None
return padded[:-pad]
except Exception: # noqa: BLE001
return None
def _decrypt_private(encoded: str, password: Optional[str]) -> Optional[Any]:
try:
data = base64.b64decode(encoded)
except Exception: # noqa: BLE001
return None
# Modern XCA: PKCS#8 EncryptedPrivateKeyInfo.
try:
return serialization.load_der_private_key(
data, password=password.encode() if password else None
)
except Exception: # noqa: BLE001
pass
# Pre-2.0 legacy: 8-byte IV + 3DES-CBC(traditional DER private key).
legacy = _legacy_decrypt(data, password)
if legacy:
try:
return serialization.load_der_private_key(legacy, password=None)
except Exception: # noqa: BLE001
pass
return None
def _item_extra(row: sqlite3.Row, item_id: int) -> Dict[str, Any]:
return {
"xca_item_id": item_id,
"source": _PKI_SOURCE.get(int(row["source"] or 0), "unknown"),
"xca_date": _parse_date(row["date"]),
"comment": str(row["comment"] or ""),
}
def import_xca_database(
data: bytes,
filename: str = "",
password: Optional[str] = None,
) -> List[ParsedItem]:
with tempfile.NamedTemporaryFile(suffix=".xdb", delete=False) as handle:
handle.write(data)
path = handle.name
try:
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
try:
schema_row = connection.execute(
"SELECT value FROM settings WHERE key_='schema'"
).fetchone()
if schema_row is None:
raise ValueError("not an XCA database (missing schema setting)")
schema = int(str(schema_row[0]))
if schema < 7:
raise ValueError(f"unsupported XCA database schema {schema}")
return _read_database(connection, filename, password)
finally:
connection.close()
except sqlite3.DatabaseError as exc:
raise ValueError("not a valid SQLite XCA database") from exc
finally:
try:
import os
os.unlink(path)
except OSError:
pass
def _read_database(
connection: sqlite3.Connection,
filename: str,
password: Optional[str],
) -> List[ParsedItem]:
items: Dict[int, sqlite3.Row] = {
int(row["id"]): row
for row in connection.execute(
"SELECT id, name, type, source, date, comment FROM items WHERE del=0"
)
}
if not items:
return []
# XCA's x509super table maps certificates/requests to key item ids.
key_refs: Dict[int, int] = {}
try:
for row in connection.execute("SELECT item, pkey FROM x509super"):
if row["pkey"] is not None:
key_refs[int(row["item"])] = int(row["pkey"])
except sqlite3.DatabaseError:
pass
result: List[ParsedItem] = []
key_objects: Dict[int, Any] = {}
# --- keys --------------------------------------------------------------
public_rows = {int(row["item"]): row for row in connection.execute("SELECT item, type, len, \"public\" FROM public_keys")}
private_rows = {int(row["item"]): row for row in connection.execute("SELECT item, ownPass, private FROM private_keys")}
for item_id, row in items.items():
if int(row["type"]) != 1 or item_id not in public_rows:
continue
public_row = public_rows[item_id]
try:
public_key = serialization.load_der_public_key(base64.b64decode(public_row["public"]))
except Exception: # noqa: BLE001
continue
key = public_key
private_row = private_rows.get(item_id)
if private_row is not None:
own_pass = int(private_row["ownPass"] or 0)
# ptCommon (0) and ptPrivate (1) are both tried with the supplied
# database password; ptBogus/ptPin cannot be used here.
if own_pass in (0, 1):
decrypted = _decrypt_private(str(private_row["private"]), password)
if decrypted is not None and is_private_key(decrypted):
key = decrypted
item = ParsedItem(
kind="key",
name=str(row["name"] or "key"),
summary=f"{public_row['type']} {public_row['len']}",
obj=key,
is_private=is_private_key(key),
extra=_item_extra(row, item_id),
)
item.extra["has_private"] = is_private_key(key)
result.append(item)
key_objects[item_id] = key
# --- certificates ------------------------------------------------------
cert_rows = {int(row["item"]): row for row in connection.execute("SELECT item, issuer, ca, cert FROM certs")}
authority_rows = {int(row["item"]): row for row in connection.execute("SELECT item, template, crlExpire, crlNo, crlDays FROM authority")}
try:
takey_rows = {int(row["item"]): row for row in connection.execute("SELECT item, value FROM takeys")}
except sqlite3.DatabaseError:
takey_rows = {}
revocations: Dict[int, List[Dict[str, Any]]] = {}
try:
for row in connection.execute(
"SELECT caId, serial, date, invaldate, crlNo, reasonBit FROM revocations"
):
revocations.setdefault(int(row["caId"]), []).append(
{
"serial_hex": str(row["serial"] or ""),
"revoked_at": _parse_date(row["date"]),
"invalidity_date": _parse_date(row["invaldate"]),
"crl_number": int(row["crlNo"] or 0),
"reason_bit": int(row["reasonBit"] or 0),
}
)
except sqlite3.DatabaseError:
pass
for item_id, row in items.items():
if int(row["type"]) != 3 or item_id not in cert_rows:
continue
cert_row = cert_rows[item_id]
try:
cert = parse_certificate(base64.b64decode(cert_row["cert"]))
except Exception: # noqa: BLE001
continue
item = ParsedItem(
kind="certificate",
name=str(row["name"] or "certificate"),
summary=cert.subject.rfc4514_string(),
obj=cert,
extra=_item_extra(row, item_id),
)
item.extra["xca_issuer_id"] = int(cert_row["issuer"]) if cert_row["issuer"] is not None else None
item.extra["xca_key_id"] = key_refs.get(item_id)
item.extra["serial_hex"] = format(cert.serial_number, "x")
authority = authority_rows.get(item_id)
if authority is not None:
item.extra["authority"] = {
"crl_expire": _parse_date(authority["crlExpire"]),
"crl_number": int(authority["crlNo"] or 0),
"crl_days": int(authority["crlDays"] or 30),
}
if item_id in takey_rows:
try:
item.extra["tls_auth_key"] = base64.b64decode(takey_rows[item_id]["value"])
except Exception: # noqa: BLE001
pass
if item_id in revocations:
item.extra["revocations"] = revocations[item_id]
result.append(item)
# --- requests ----------------------------------------------------------
request_rows = {int(row["item"]): row for row in connection.execute("SELECT item, signed, request FROM requests")}
for item_id, row in items.items():
if int(row["type"]) != 2 or item_id not in request_rows:
continue
request_row = request_rows[item_id]
try:
csr = parse_csr(base64.b64decode(request_row["request"]))
except Exception: # noqa: BLE001
continue
item = ParsedItem(
kind="request",
name=str(row["name"] or "request"),
summary=csr.subject.rfc4514_string(),
obj=csr,
extra=_item_extra(row, item_id),
)
item.extra["xca_key_id"] = key_refs.get(item_id)
item.extra["signed"] = bool(request_row["signed"])
result.append(item)
# --- CRLs --------------------------------------------------------------
crl_rows = {int(row["item"]): row for row in connection.execute("SELECT item, issuer, crl FROM crls")}
for item_id, row in items.items():
if int(row["type"]) != 4 or item_id not in crl_rows:
continue
crl_row = crl_rows[item_id]
try:
crl = parse_crl(base64.b64decode(crl_row["crl"]))
except Exception: # noqa: BLE001
continue
item = ParsedItem(
kind="crl",
name=str(row["name"] or "crl"),
summary=crl.issuer.rfc4514_string(),
obj=crl,
extra=_item_extra(row, item_id),
)
item.extra["xca_issuer_id"] = int(crl_row["issuer"]) if crl_row["issuer"] is not None else None
result.append(item)
# --- templates ---------------------------------------------------------
template_rows = {int(row["item"]): row for row in connection.execute("SELECT item, version, template FROM templates")}
for item_id, row in items.items():
if int(row["type"]) != 5 or item_id not in template_rows:
continue
template_row = template_rows[item_id]
try:
payload = base64.b64decode(template_row["template"])
template = template_from_xca_payload(payload, int(template_row["version"] or 10))
except Exception: # noqa: BLE001
continue
template["name"] = str(row["name"] or template.get("name") or "template")
template["comment"] = str(row["comment"] or "")
item = ParsedItem(
kind="template",
name=str(row["name"] or "template"),
summary=str(row["comment"] or ""),
obj=template,
extra=_item_extra(row, item_id),
)
result.append(item)
return result
__all__ = ["import_xca_database"]
+302
View File
@@ -0,0 +1,302 @@
"""Core domain tests: names, extensions, keys, certificates, CRLs and exports."""
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from kca.core import certificates, crls, crypto, exports, imports, keys, names, requests, templates
from kca.core.extensions import (
build_extensions,
default_extension_spec,
extension_specs,
)
def _name(cn: str = "Test CA") -> x509.Name:
return names.name_from_entries(
names.entries_from_json(
[
{"name": "C", "value": "DE"},
{"name": "O", "value": "KCA"},
{"name": "CN", "value": cn},
]
)
)
def _ca_spec() -> dict:
spec = default_extension_spec()
spec["basic_constraints"] = {"critical": True, "ca": True, "path_length": None}
spec["key_usage"] = {"critical": True, "usages": ["key_cert_sign", "crl_sign"]}
spec["subject_key_identifier"] = {"critical": False}
spec["authority_key_identifier"] = {"critical": False}
return spec
class NamesTest(unittest.TestCase):
def test_order_and_duplicates_survive(self) -> None:
name = _name()
entries = names.entries_from_name(name)
self.assertEqual([e.name for e in entries], ["C", "O", "CN"])
self.assertEqual(name.rfc4514_string(), "CN=Test CA,O=KCA,C=DE")
def test_der_roundtrip(self) -> None:
name = _name()
self.assertEqual(names.name_from_der(names.name_to_der(name)), name)
class ExtensionTest(unittest.TestCase):
def test_roundtrip(self) -> None:
key = keys.generate_key("rsa", 2048)
spec = _ca_spec()
spec["subject_alt_name"] = {
"critical": False,
"entries": [
{"type": "dns", "value": "example.com"},
{"type": "ip", "value": "192.0.2.1"},
{"type": "email", "value": "ca@example.com"},
],
}
exts = build_extensions(spec, public_key=key.public_key(), subject_name=_name())
parsed = extension_specs(exts)
self.assertTrue(parsed["basic_constraints"]["ca"])
self.assertEqual(parsed["basic_constraints"]["path_length"], None)
self.assertEqual(len(parsed["subject_alt_name"]["entries"]), 3)
self.assertIn("key_cert_sign", parsed["key_usage"]["usages"])
def test_raw_extension(self) -> None:
key = keys.generate_key("ec", curve="secp256r1")
spec = default_extension_spec()
spec["raw"] = [{"oid": "1.2.3.4", "critical": True, "der_hex": "0c03616263"}]
exts = build_extensions(spec, public_key=key.public_key(), subject_name=_name())
self.assertEqual(exts[0].oid.dotted_string, "1.2.3.4")
self.assertTrue(exts[0].critical)
class KeyTest(unittest.TestCase):
def test_all_types(self) -> None:
for key_type, kwargs in [
("rsa", {"size": 2048}),
("ec", {"curve": "secp256r1"}),
("ed25519", {}),
]:
key = keys.generate_key(key_type, **kwargs)
info = keys.key_info(key)
self.assertTrue(info.is_private)
self.assertEqual(info.key_type.upper() if key_type != "ed25519" else "ED25519", info.key_type)
loaded, private = keys.load_key_bytes(keys.private_pem(key))
self.assertTrue(private)
self.assertEqual(keys.public_der(loaded), info.public_der)
def test_encrypted_export_roundtrip(self) -> None:
key = keys.generate_key("rsa", 2048)
encrypted = keys.private_pem(key, password="secret")
loaded, private = keys.load_key_bytes(encrypted, password="secret")
self.assertTrue(private)
self.assertEqual(keys.public_der(loaded), keys.public_der(key))
def test_ssh_and_pvk(self) -> None:
key = keys.generate_key("rsa", 2048)
public = keys.ssh_public(key, "test")
loaded, private = keys.load_key_bytes(public.encode())
self.assertFalse(private)
pvk = keys.pvk_encode_private(key)
loaded, private = keys.load_key_bytes(pvk)
self.assertTrue(private)
self.assertEqual(keys.public_der(loaded), keys.public_der(key))
class CertificateTest(unittest.TestCase):
def test_issue_and_verify(self) -> None:
ca_key = keys.generate_key("rsa", 2048)
now = datetime.now(timezone.utc)
ca = certificates.build_self_signed(
subject=_name("Root CA"),
private_key=ca_key,
serial=1,
not_before=now,
not_after=now + timedelta(days=3650),
extension_spec=_ca_spec(),
)
self.assertTrue(certificates.is_self_signed(ca))
self.assertTrue(certificates.certificate_info(ca).is_ca)
leaf_key = keys.generate_key("ec", curve="secp256r1")
spec = default_extension_spec()
spec["basic_constraints"] = {"critical": True, "ca": False, "path_length": None}
spec["key_usage"] = {"critical": True, "usages": ["digital_signature", "key_encipherment"]}
spec["extended_key_usage"] = {"critical": False, "usages": ["serverAuth"]}
spec["subject_alt_name"] = {"critical": False, "entries": [{"type": "dns", "value": "example.com"}]}
leaf = certificates.build_signed(
subject=_name("example.com"),
public_key=leaf_key.public_key(),
issuer_cert=ca,
issuer_private_key=ca_key,
serial=2,
not_before=now,
not_after=now + timedelta(days=365),
extension_spec=spec,
)
self.assertTrue(certificates.verify_directly_issued_by(leaf, ca))
self.assertFalse(certificates.verify_directly_issued_by(leaf, leaf))
self.assertIn("SSL server", certificates.purposes(leaf))
parsed = certificates.parse_certificate(certificates.certificate_der(leaf))
self.assertEqual(parsed.serial_number, 2)
self.assertEqual(certificates.find_issuer(parsed, [ca]), ca)
self.assertEqual(certificates.validation_errors(parsed, [parsed, ca]), [])
def test_renew_spec(self) -> None:
key = keys.generate_key("rsa", 2048)
now = datetime.now(timezone.utc)
cert = certificates.build_self_signed(
subject=_name("Renew Me"),
private_key=key,
serial=3,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
)
spec = certificates.certificate_spec(cert)
self.assertEqual(spec["serial_hex"], "3")
self.assertEqual(spec["subject"][-1]["name"], "CN")
def test_index_and_config(self) -> None:
key = keys.generate_key("rsa", 2048)
now = datetime.now(timezone.utc)
cert = certificates.build_self_signed(
subject=_name("Index"),
private_key=key,
serial=4,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
)
self.assertIn("V\t", certificates.index_entry(cert))
self.assertIn("basicConstraints", certificates.openssl_config(cert))
self.assertIn("VCALENDAR", certificates.ics_calendar([certificates.ics_event(cert, "Index")]))
class RequestTest(unittest.TestCase):
def test_csr_roundtrip(self) -> None:
key = keys.generate_key("rsa", 2048)
csr = requests.build_csr(
subject=_name("CSR"),
private_key=key,
extension_spec=_ca_spec(),
attributes={requests.CHALLENGE_PASSWORD_OID: "secret"},
)
self.assertTrue(requests.verify_csr(csr))
loaded = requests.parse_csr(requests.csr_der(csr))
self.assertEqual(requests.request_info(loaded).subject, csr.subject)
self.assertEqual(requests.attribute_values(loaded)[requests.CHALLENGE_PASSWORD_OID], "secret")
class CrlTest(unittest.TestCase):
def test_build_and_parse(self) -> None:
key = keys.generate_key("rsa", 2048)
now = datetime.now(timezone.utc)
ca = certificates.build_self_signed(
subject=_name("CRL CA"),
private_key=key,
serial=5,
not_before=now,
not_after=now + timedelta(days=3650),
extension_spec=_ca_spec(),
)
crl = crls.build_crl(
issuer_cert=ca,
issuer_private_key=key,
revocations=[
{
"serial_hex": "10",
"revoked_at": now,
"reason": "keyCompromise",
"invalidity_date": now - timedelta(days=1),
}
],
last_update=now,
next_update=now + timedelta(days=30),
crl_number=7,
)
info = crls.crl_info(crl)
self.assertEqual(info.crl_number, 7)
self.assertEqual(info.revoked_count, 1)
self.assertTrue(crls.verify_crl(crl, ca))
parsed = crls.parse_crl(crls.crl_der(crl))
rows = crls.crl_revocations(parsed)
self.assertEqual(rows[0]["serial_hex"], "10")
self.assertEqual(rows[0]["reason"], "keyCompromise")
class TemplateTest(unittest.TestCase):
def test_xca_roundtrip(self) -> None:
template = templates.builtin_templates()[0].as_dict()
data = templates.template_to_xca_data(template)
parsed = templates.template_from_xca_data(data)
self.assertEqual(parsed["name"], "CA")
self.assertTrue(parsed["settings"]["extensions"]["basic_constraints"]["ca"])
def test_apply(self) -> None:
template = templates.builtin_templates()[1].as_dict()
merged = templates.apply_template(template, {"subject": [], "extensions": {}})
self.assertIn("serverAuth", merged["extensions"]["extended_key_usage"]["usages"])
class ExportImportTest(unittest.TestCase):
def setUp(self) -> None:
self.key = keys.generate_key("rsa", 2048)
now = datetime.now(timezone.utc)
self.cert = certificates.build_self_signed(
subject=_name("Export"),
private_key=self.key,
serial=6,
not_before=now,
not_after=now + timedelta(days=30),
extension_spec=_ca_spec(),
)
def test_certificate_exports(self) -> None:
for fmt, prefix in [
("pem", b"-----BEGIN CERTIFICATE"),
("der", b"0"),
("pkcs7", b"0"),
("pkcs12", b"0"),
]:
result = exports.export_certificate(
self.cert, fmt, name="export", private_key=self.key, password="secret"
)
self.assertTrue(result.data.startswith(prefix), fmt)
pem = exports.export_certificate(self.cert, "pem", name="export")
items = imports.parse_file(pem.data, "export.crt")
self.assertEqual(items[0].kind, "certificate")
def test_key_exports(self) -> None:
for fmt in ("pem-public", "pem-private", "pkcs8", "ssh-public", "jwk-public", "pvk"):
result = exports.export_key(self.key, fmt, name="export")
self.assertTrue(result.data)
encrypted = exports.export_key(self.key, "pkcs8-encrypted", name="export", password="secret")
loaded, private = keys.load_key_bytes(encrypted.data, password="secret")
self.assertTrue(private)
def test_pkcs12_contains_chain(self) -> None:
result = exports.export_certificate(
self.cert, "pkcs12-chain", name="export", chain=[self.cert], private_key=self.key, password="secret"
)
items = imports.parse_file(result.data, "export.pfx", "secret")
self.assertGreaterEqual(len(items), 2)
class CryptoTest(unittest.TestCase):
def test_envelope(self) -> None:
master = crypto.generate_master_key()
blob = crypto.encrypt_private_key(b"secret", "key-id", crypto.decode_master_key(master))
self.assertEqual(crypto.decrypt_private_key(blob, "key-id", crypto.decode_master_key(master)), b"secret")
with self.assertRaises(crypto.DecryptionError):
crypto.decrypt_private_key(blob, "other-id", crypto.decode_master_key(master))
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,135 @@
"""XCA .xdb import tests using a synthetic schema-compatible database."""
from __future__ import annotations
import base64
import sqlite3
import struct
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from cryptography.hazmat.primitives import serialization
from kca.core import certificates, crls, keys, names, requests, templates
from kca.core.extensions import build_extensions, default_extension_spec
from kca.core.xca_db import import_xca_database
def _name(cn: str) -> object:
return names.name_from_entries(names.entries_from_json([{"name": "CN", "value": cn}]))
class XcaDbTest(unittest.TestCase):
def test_import_synthetic_database(self) -> None:
key = keys.generate_key("rsa", 2048)
now = datetime.now(timezone.utc)
spec = default_extension_spec()
spec["basic_constraints"] = {"critical": True, "ca": True, "path_length": None}
exts = build_extensions(spec, public_key=key.public_key(), subject_name=_name("Synthetic CA"))
cert = certificates.build_certificate(
subject=_name("Synthetic CA"),
public_key=key.public_key(),
issuer_cert=None,
issuer_private_key=key,
serial=42,
not_before=now,
not_after=now + timedelta(days=365),
extensions=exts,
)
csr = requests.build_csr(subject=_name("Synthetic CSR"), private_key=key, extension_spec=spec)
crl = crls.build_crl(
issuer_cert=cert,
issuer_private_key=key,
revocations=[{"serial_hex": "2a", "revoked_at": now, "reason": "keyCompromise"}],
last_update=now,
next_update=now + timedelta(days=30),
crl_number=3,
)
tmpl = templates.builtin_templates()[0].as_dict()
tmpl_pem = templates.template_to_xca_data(tmpl)
body = tmpl_pem.decode().split("-----BEGIN XCA TEMPLATE-----")[1].split("-----END XCA TEMPLATE-----")[0]
payload = base64.b64decode("".join(body.split()))
tmpl_data = payload[8:]
date = now.strftime("%Y%m%d%H%M%SZ")
with tempfile.NamedTemporaryFile(suffix=".xdb", delete=False) as handle:
path = handle.name
try:
db = sqlite3.connect(path)
db.executescript(
"""
CREATE TABLE settings (key_ CHAR(20) UNIQUE, value TEXT);
CREATE TABLE items (id INTEGER PRIMARY KEY, name VARCHAR(128), type INTEGER, source INTEGER, date CHAR(15), comment VARCHAR(2048), stamp INTEGER DEFAULT 0, del SMALLINT DEFAULT 0);
CREATE TABLE public_keys (item INTEGER, type CHAR(4), hash INTEGER, len INTEGER, "public" TEXT);
CREATE TABLE private_keys (item INTEGER, ownPass INTEGER, private TEXT);
CREATE TABLE x509super (item INTEGER, subj_hash INTEGER, pkey INTEGER, key_hash INTEGER);
CREATE TABLE requests (item INTEGER, hash INTEGER, signed INTEGER, request TEXT);
CREATE TABLE certs (item INTEGER, hash INTEGER, iss_hash INTEGER, serial VARCHAR(64), issuer INTEGER, ca INTEGER, cert TEXT);
CREATE TABLE authority (item INTEGER, template INTEGER, crlExpire CHAR(15), crlNo INTEGER, crlDays INTEGER);
CREATE TABLE crls (item INTEGER, hash INTEGER, num INTEGER, iss_hash INTEGER, issuer INTEGER, crl TEXT);
CREATE TABLE revocations (caId INTEGER, serial VARCHAR(64), date CHAR(15), invaldate CHAR(15), crlNo INTEGER, reasonBit INTEGER);
CREATE TABLE templates (item INTEGER, version INTEGER, template TEXT);
CREATE TABLE takeys (item INTEGER UNIQUE, value TEXT);
"""
)
db.execute("INSERT INTO settings VALUES ('schema', '8')")
db.execute("INSERT INTO items VALUES (1, 'Synthetic key', 1, 2, ?, '', 0, 0)", (date,))
db.execute("INSERT INTO items VALUES (2, 'Synthetic CA', 3, 2, ?, '', 0, 0)", (date,))
db.execute("INSERT INTO items VALUES (3, 'Synthetic CSR', 2, 2, ?, '', 0, 0)", (date,))
db.execute("INSERT INTO items VALUES (4, 'Synthetic CRL', 4, 2, ?, '', 0, 0)", (date,))
db.execute("INSERT INTO items VALUES (5, 'Synthetic template', 5, 2, ?, '', 0, 0)", (date,))
encrypted = key.private_bytes(
serialization.Encoding.DER,
serialization.PrivateFormat.PKCS8,
serialization.BestAvailableEncryption(b"dbpass"),
)
db.execute(
"INSERT INTO public_keys VALUES (1, 'RSA', 0, 2048, ?)",
(base64.b64encode(keys.public_der(key)).decode(),),
)
db.execute(
"INSERT INTO private_keys VALUES (1, 0, ?)",
(base64.b64encode(encrypted).decode(),),
)
db.execute("INSERT INTO x509super VALUES (2, 0, 1, 0)")
db.execute("INSERT INTO x509super VALUES (3, 0, 1, 0)")
db.execute(
"INSERT INTO certs VALUES (2, 0, 0, '2a', NULL, 1, ?)",
(base64.b64encode(certificates.certificate_der(cert)).decode(),),
)
db.execute("INSERT INTO authority VALUES (2, NULL, NULL, 0, 30)")
db.execute(
"INSERT INTO revocations VALUES (2, '2a', ?, NULL, 0, 1)",
(date,),
)
db.execute(
"INSERT INTO requests VALUES (3, 0, 0, ?)",
(base64.b64encode(requests.csr_der(csr)).decode(),),
)
db.execute(
"INSERT INTO crls VALUES (4, 0, 1, 0, 2, ?)",
(base64.b64encode(crls.crl_der(crl)).decode(),),
)
db.execute(
"INSERT INTO templates VALUES (5, 10, ?)",
(base64.b64encode(tmpl_data).decode(),),
)
db.commit()
db.close()
with open(path, "rb") as handle:
items = import_xca_database(handle.read(), "synthetic.xdb", "dbpass")
finally:
import os
os.unlink(path)
kinds = [item.kind for item in items]
self.assertEqual(kinds, ["key", "certificate", "request", "crl", "template"])
self.assertTrue(items[0].is_private)
self.assertEqual(items[1].name, "Synthetic CA")
self.assertEqual(items[3].name, "Synthetic CRL")
self.assertEqual(items[4].obj["name"], "Synthetic template")
if __name__ == "__main__":
unittest.main()