# 29. `example_crmf.py` — CRMF certificate request messages
[← Example index](index.md) · [example_crmf.py on Codeberg](https://codeberg.org/abbra/synta/src/branch/main/examples/example_crmf.py)
Bindings: `CertReqMsgBuilder`, `CertReqMessagesBuilder`, `CertReqMessages.from_der`,
`CertReqMsg.from_der`; OID constants from `synta.crmf`.
- Build a `CertReqMsg` with `CertReqMsgBuilder`: set subject name via `NameBuilder`,
add a `SubjectPublicKeyInfo` DER, set validity, and call `build()`.
- Assemble a batch with `CertReqMessagesBuilder.add(msg).build()`.
- DER round-trip via `CertReqMessages.from_der`; iterate the batch and inspect
each request's `cert_req_id`, subject, and public key fields.
- Print OID constants from `synta.crmf`.
## Source
```python
#!/usr/bin/env python3
"""
Example: RFC 4211 Certificate Request Message Format (CRMF).
Demonstrates:
- Building a CertReqMsg with the fluent CertReqMsgBuilder
- Assembling a CertReqMessages batch with CertReqMessagesBuilder
- DER encoding and round-trip parsing via CertReqMessages.from_der
- Iterating over the batch and inspecting individual request fields
- OID constants exported by synta.crmf
"""
import synta
import synta.crmf as crmf
def section(title):
print(f"\n{'─' * 60}\n{title}\n{'─' * 60}")
def demo_names():
section("1. Build subject and issuer Names")
subject_der = (
synta.NameBuilder()
.common_name("host.example.com")
.organization("Example Org")
.build()
)
print(f"Subject Name DER ({len(subject_der)} bytes): {subject_der.hex()}")
issuer_der = (
synta.NameBuilder()
.common_name("Example CA")
.organization("Example Org")
.build()
)
print(f"Issuer Name DER ({len(issuer_der)} bytes): {issuer_der.hex()}")
return subject_der, issuer_der
def demo_build_requests(subject_der, issuer_der):
section("2. Build CertReqMsg objects")
# Request 0: subject only, raVerified proof-of-possession
req0 = (
crmf.CertReqMsgBuilder()
.cert_req_id(0)
.subject_name(subject_der)
.popo_ra_verified()
.build()
)
print(f"req0 cert_req_id : {req0.cert_req_id}")
print(f"req0 popo_type : {req0.popo_type}")
print(f"req0 subject_der : {req0.subject_der.hex() if req0.subject_der else None}")
print(f"req0 issuer_der : {req0.issuer_der}")
# Request 1: subject + issuer, no popo
req1 = (
crmf.CertReqMsgBuilder()
.cert_req_id(1)
.subject_name(subject_der)
.issuer_name(issuer_der)
.build()
)
print(f"\nreq1 cert_req_id : {req1.cert_req_id}")
print(f"req1 popo_type : {req1.popo_type}")
print(f"req1 issuer_der : {req1.issuer_der.hex() if req1.issuer_der else None}")
return req0, req1
def demo_batch(req0, req1):
section("3. Assemble CertReqMessages batch")
msgs = (
crmf.CertReqMessagesBuilder()
.add_request(req0)
.add_request(req1)
.build()
)
print(f"Batch: {repr(msgs)}")
print(f"Count: {len(msgs)}")
return msgs
def demo_der_encoding(msgs):
section("4. DER encoding")
der = msgs.to_der()
print(f"DER length: {len(der)} bytes")
print(f"DER (hex): {der.hex()}")
return der
def demo_roundtrip(der, subject_der, issuer_der):
section("5. Round-trip parse via CertReqMessages.from_der")
msgs2 = crmf.CertReqMessages.from_der(der)
print(f"Parsed: {repr(msgs2)}")
for req in msgs2:
print(
f" id={req.cert_req_id} popo={req.popo_type} "
f"subject={'yes' if req.subject_der else 'no'} "
f"issuer={'yes' if req.issuer_der else 'no'}"
)
assert len(msgs2) == 2
assert msgs2[0].cert_req_id == 0
assert msgs2[0].popo_type == "raVerified"
assert msgs2[0].subject_der == subject_der
assert msgs2[0].issuer_der is None
assert msgs2[1].cert_req_id == 1
assert msgs2[1].popo_type is None
assert msgs2[1].issuer_der == issuer_der
print("All assertions passed.")
return msgs2
def demo_indexing(msgs2):
section("6. Indexing and iteration")
r = msgs2[0]
print(f"msgs2[0]: {repr(r)}")
print(f"cert_template_der ({len(r.cert_template_der)} bytes): "
f"{r.cert_template_der.hex()}")
for i, req in enumerate(msgs2):
print(f" [{i}] id={req.cert_req_id} popo_der="
f"{'None' if req.popo_der is None else req.popo_der.hex()}")
def demo_oid_constants():
section("7. Registration-control OID constants")
oids = [
("ID_REG_CTRL_REG_TOKEN", crmf.ID_REG_CTRL_REG_TOKEN),
("ID_REG_CTRL_AUTHENTICATOR", crmf.ID_REG_CTRL_AUTHENTICATOR),
("ID_REG_CTRL_PKI_PUBLICATION_INFO", crmf.ID_REG_CTRL_PKI_PUBLICATION_INFO),
("ID_REG_CTRL_PKI_ARCHIVE_OPTIONS", crmf.ID_REG_CTRL_PKI_ARCHIVE_OPTIONS),
("ID_REG_CTRL_OLD_CERT_ID", crmf.ID_REG_CTRL_OLD_CERT_ID),
("ID_REG_CTRL_PROTOCOL_ENCR_KEY", crmf.ID_REG_CTRL_PROTOCOL_ENCR_KEY),
("ID_REG_INFO_UTF8_PAIRS", crmf.ID_REG_INFO_UTF8_PAIRS),
("ID_REG_INFO_CERT_REQ", crmf.ID_REG_INFO_CERT_REQ),
]
for name, oid in oids:
print(f" {name:40s} = {oid}")
def main():
print("=" * 60)
print("Example: CRMF (RFC 4211)")
print("=" * 60)
subject_der, issuer_der = demo_names()
req0, req1 = demo_build_requests(subject_der, issuer_der)
msgs = demo_batch(req0, req1)
der = demo_der_encoding(msgs)
msgs2 = demo_roundtrip(der, subject_der, issuer_der)
demo_indexing(msgs2)
demo_oid_constants()
print("\nDone.")
if __name__ == "__main__":
main()
```