synta 0.1.9

#!/usr/bin/env python3
"""
Example 20: RFC 5755 Attribute Certificate (AC) parsing.

Demonstrates: synta.ac.AttributeCertificate.from_der, all getters
(version, serial_number, not_before, not_after, signature_algorithm_oid,
signature, holder_der, issuer_der, attributes_der), to_der round-trip,
and OID constants exported by synta.ac.
"""

import synta
import synta.ac as ac


def section(title):
    print(f"\n{'─' * 60}\n{title}\n{'─' * 60}")


# ── DER building helpers ──────────────────────────────────────

def _enc(fn):
    """Encode one element and return its DER bytes."""
    enc = synta.Encoder(synta.Encoding.DER)
    fn(enc)
    return enc.finish()


def _seq(*parts):
    """Wrap raw TLV bytes in a SEQUENCE (tag 0x30)."""
    enc = synta.Encoder(synta.Encoding.DER)
    enc.encode_sequence(b"".join(parts))
    return enc.finish()


def _explicit(n, content):
    """Wrap bytes in a [n] EXPLICIT context tag (constructed)."""
    enc = synta.Encoder(synta.Encoding.DER)
    enc.encode_explicit_tag(n, "Context", content)
    return enc.finish()


def _implicit(n, inner_der):
    """[n] IMPLICIT: replace the original tag byte with a context tag.

    Preserves the constructed bit.  Only short-form length (<128) is
    handled here, which is sufficient for test vectors.
    """
    orig_tag = inner_der[0]
    constructed = orig_tag & 0x20
    length_byte = inner_der[1]
    assert length_byte < 0x80, "only short-form length supported in helper"
    content = inner_der[2:2 + length_byte]
    ctx_tag = 0x80 | constructed | n
    return bytes([ctx_tag, len(content)]) + content


def _int_der(v):
    return _enc(lambda e: e.encode_integer(v))


def _oct_der(b):
    return _enc(lambda e: e.encode_octet_string(b))


def _oid_der(s):
    return _enc(lambda e: e.encode_oid(synta.ObjectIdentifier(s)))


def _bit_der(b):
    return _enc(lambda e: e.encode_bit_string(synta.BitString(b, 0)))


def _gt_der(s):
    """Encode a GeneralizedTime string (YYYYMMDDHHmmssZ)."""
    year   = int(s[0:4])
    month  = int(s[4:6])
    day    = int(s[6:8])
    hour   = int(s[8:10])
    minute = int(s[10:12])
    second = int(s[12:14])
    return _enc(
        lambda e: e.encode_generalized_time(
            synta.GeneralizedTime(year, month, day, hour, minute, second, None)
        )
    )


# ── Build a minimal but parseable AttributeCertificate v2 ────
#
# RFC 5755 §4.1 structure:
#
#  AttributeCertificate ::= SEQUENCE {
#    acinfo             AttributeCertificateInfo,
#    signatureAlgorithm AlgorithmIdentifier,
#    signature          BIT STRING }
#
#  AttributeCertificateInfo ::= SEQUENCE {
#    version           INTEGER (v2 = 1),
#    holder            Holder,
#    issuer            AttCertIssuer (CHOICE),
#    signature         AlgorithmIdentifier,
#    serialNumber      INTEGER,
#    attrCertValidityPeriod AttCertValidityPeriod,
#    attributes        SEQUENCE OF Attribute,
#    ... }
#
# We use the simplest legal encoding for each optional sub-structure:
#   holder   — empty SEQUENCE (all three optional fields absent)
#   issuer   — v1Form: empty GeneralNames SEQUENCE (tag 0x30, length 0x00)
#              (v1Form has no context tag in the CHOICE)
#   AlgorithmIdentifier — SEQUENCE { OID sha256WithRSAEncryption, NULL }
#   attributes — empty SEQUENCE (no attributes)

# sha256WithRSAEncryption (1.2.840.113549.1.1.11)
SHA256_WITH_RSA_OID = "1.2.840.113549.1.1.11"

def _alg_id(oid_str):
    """AlgorithmIdentifier ::= SEQUENCE { algorithm OID, parameters NULL OPTIONAL }."""
    null_der = _enc(lambda e: e.encode_null())
    return _seq(_oid_der(oid_str), null_der)


# Holder: all three fields (baseCertificateID, entityName, objectDigestInfo)
# are OPTIONAL — an empty SEQUENCE is a valid Holder.
HOLDER_DER = _seq()  # SEQUENCE {}

# AttCertIssuer CHOICE v1Form = GeneralNames = SEQUENCE OF GeneralName.
# An empty SEQUENCE is a valid (though degenerate) GeneralNames value.
# v1Form has no context tag; the SEQUENCE tag 0x30 discriminates the CHOICE.
ISSUER_DER = _seq()  # SEQUENCE {} — v1Form with zero GeneralNames

# AlgorithmIdentifier used both inside acinfo.signature and outer signatureAlgorithm
ALG_ID_DER = _alg_id(SHA256_WITH_RSA_OID)

# attrCertValidityPeriod ::= SEQUENCE { notBeforeTime GeneralizedTime,
#                                        notAfterTime  GeneralizedTime }
NOT_BEFORE = "20260101000000Z"
NOT_AFTER  = "20370101000000Z"
VALIDITY_DER = _seq(_gt_der(NOT_BEFORE), _gt_der(NOT_AFTER))

# attributes ::= SEQUENCE OF Attribute  (empty list is valid)
ATTRIBUTES_DER = _seq()

# serialNumber: INTEGER 42
SERIAL_DER = _int_der(42)

# version: INTEGER 1  (AttCertVersion v2)
VERSION_DER = _int_der(1)

# AttributeCertificateInfo
ACINFO_DER = _seq(
    VERSION_DER,
    HOLDER_DER,
    ISSUER_DER,
    ALG_ID_DER,    # acinfo.signature (AlgorithmIdentifier)
    SERIAL_DER,
    VALIDITY_DER,
    ATTRIBUTES_DER,
)

# Outer signature: 4-byte dummy RSA signature blob
SIGNATURE_DER = _bit_der(b"\xaa\xbb\xcc\xdd")

# Full AttributeCertificate
AC_DER = _seq(ACINFO_DER, ALG_ID_DER, SIGNATURE_DER)


# ── Demo functions ────────────────────────────────────────────

def demo_parse_and_getters():
    section("AttributeCertificate.from_der — all getters")
    acer = ac.AttributeCertificate.from_der(AC_DER)

    # version: always 1 for v2 AC (RFC 5755)
    assert acer.version == 1
    print(f"  version:                {acer.version}  (1 = v2)")

    # serial_number: big-endian bytes of INTEGER 42
    sn = acer.serial_number
    assert isinstance(sn, bytes)
    assert int.from_bytes(sn, "big") == 42
    print(f"  serial_number:          {sn.hex()}  ({int.from_bytes(sn, 'big')})")

    # Validity period strings
    nb = acer.not_before
    na = acer.not_after
    assert NOT_BEFORE in nb
    assert NOT_AFTER  in na
    print(f"  not_before:             {nb}")
    print(f"  not_after:              {na}")

    # Signature algorithm OID (from acinfo.signature, not the outer field)
    sig_oid = acer.signature_algorithm_oid
    assert str(sig_oid) == SHA256_WITH_RSA_OID
    print(f"  signature_algorithm_oid: {sig_oid}")

    # Raw signature bytes (BitString value, no unused-bits prefix)
    sig = acer.signature
    assert sig == b"\xaa\xbb\xcc\xdd"
    print(f"  signature:              {sig.hex()}  ({len(sig)} bytes)")

    # holder_der: DER encoding of the Holder SEQUENCE
    h_der = acer.holder_der
    assert isinstance(h_der, bytes)
    assert h_der[0] == 0x30, "Holder DER must start with SEQUENCE tag"
    print(f"  holder_der:             {h_der.hex()}  (Holder SEQUENCE)")

    # issuer_der: DER encoding of the AttCertIssuer CHOICE
    i_der = acer.issuer_der
    assert isinstance(i_der, bytes)
    print(f"  issuer_der:             {i_der.hex()}  (AttCertIssuer)")

    # attributes_der: DER encoding of SEQUENCE OF Attribute
    a_der = acer.attributes_der
    assert isinstance(a_der, bytes)
    assert a_der[0] == 0x30, "attributes DER must start with SEQUENCE tag"
    print(f"  attributes_der:         {a_der.hex()}  (SEQUENCE OF Attribute)")

    print(f"  repr:                   {repr(acer)}")


def demo_to_der_round_trip():
    section("to_der — round-trip fidelity")
    acer = ac.AttributeCertificate.from_der(AC_DER)
    reparsed_der = acer.to_der()
    # Re-parse the encoded form and check it decodes without error
    acer2 = ac.AttributeCertificate.from_der(reparsed_der)
    assert acer2.version == acer.version
    assert acer2.serial_number == acer.serial_number
    assert acer2.not_before == acer.not_before
    assert acer2.not_after  == acer.not_after
    assert str(acer2.signature_algorithm_oid) == str(acer.signature_algorithm_oid)
    assert acer2.signature == acer.signature
    print(f"  Original DER:  {len(AC_DER)} bytes")
    print(f"  to_der output: {len(reparsed_der)} bytes")
    print(f"  Round-trip: OK  (re-parsed version={acer2.version}, serial={int.from_bytes(acer2.serial_number, 'big')})")


def demo_oid_constants():
    section("OID constants exported by synta.ac")
    oids = {
        "ID_PE_AC_AUDIT_IDENTITY":       ac.ID_PE_AC_AUDIT_IDENTITY,
        "ID_PE_AA_CONTROLS":             ac.ID_PE_AA_CONTROLS,
        "ID_PE_AC_PROXYING":             ac.ID_PE_AC_PROXYING,
        "ID_CE_TARGET_INFORMATION":      ac.ID_CE_TARGET_INFORMATION,
        "ID_ACA_AUTHENTICATION_INFO":    ac.ID_ACA_AUTHENTICATION_INFO,
        "ID_ACA_ACCESS_IDENTITY":        ac.ID_ACA_ACCESS_IDENTITY,
        "ID_ACA_CHARGING_IDENTITY":      ac.ID_ACA_CHARGING_IDENTITY,
        "ID_ACA_GROUP":                  ac.ID_ACA_GROUP,
        "ID_ACA_ENC_ATTRS":              ac.ID_ACA_ENC_ATTRS,
        "ID_AT_ROLE":                    ac.ID_AT_ROLE,
        "ID_AT_CLEARANCE":               ac.ID_AT_CLEARANCE,
    }
    for name, oid in oids.items():
        print(f"  {name:<32} {oid}")

    # Spot-check a few known OID values from RFC 5755
    assert str(ac.ID_ACA_AUTHENTICATION_INFO) == "1.3.6.1.5.5.7.10.1"
    assert str(ac.ID_ACA_ACCESS_IDENTITY)     == "1.3.6.1.5.5.7.10.2"
    assert str(ac.ID_ACA_CHARGING_IDENTITY)   == "1.3.6.1.5.5.7.10.3"
    assert str(ac.ID_ACA_GROUP)               == "1.3.6.1.5.5.7.10.4"
    assert str(ac.ID_ACA_ENC_ATTRS)           == "1.3.6.1.5.5.7.10.6"
    assert str(ac.ID_AT_ROLE)                 == "2.5.4.72"
    assert str(ac.ID_AT_CLEARANCE)            == "2.5.4.55"
    assert str(ac.ID_CE_TARGET_INFORMATION)   == "2.5.29.55"
    assert str(ac.ID_PE_AC_AUDIT_IDENTITY)    == "1.3.6.1.5.5.7.1.4"
    assert str(ac.ID_PE_AC_PROXYING)          == "1.3.6.1.5.5.7.1.10"
    assert str(ac.ID_PE_AA_CONTROLS)          == "1.3.6.1.5.5.7.1.56"
    print("\n  All OID value assertions passed.")


def demo_inspect_holder_and_issuer():
    section("Re-decoding holder_der and issuer_der with a Decoder")
    acer = ac.AttributeCertificate.from_der(AC_DER)

    # holder_der can be re-decoded as a SEQUENCE by a Decoder
    h_dec = synta.Decoder(acer.holder_der, synta.Encoding.DER)
    h_seq = h_dec.decode_sequence()
    assert h_seq.is_empty(), "our Holder has no fields"
    print("  Holder SEQUENCE is empty (all fields absent): OK")

    # issuer_der for the v1Form CHOICE arm is itself a SEQUENCE (GeneralNames)
    i_dec = synta.Decoder(acer.issuer_der, synta.Encoding.DER)
    i_seq = i_dec.decode_sequence()
    assert i_seq.is_empty(), "our GeneralNames (v1Form) has no entries"
    print("  Issuer v1Form GeneralNames SEQUENCE is empty: OK")

    # attributes_der is a SEQUENCE OF Attribute (empty in our example)
    a_dec = synta.Decoder(acer.attributes_der, synta.Encoding.DER)
    a_seq = a_dec.decode_sequence()
    assert a_seq.is_empty(), "our SEQUENCE OF Attribute has no entries"
    print("  attributes SEQUENCE OF Attribute is empty: OK")


def main():
    print("=" * 60)
    print("Example 20: RFC 5755 Attribute Certificate (AC) parsing")
    print("=" * 60)
    demo_parse_and_getters()
    demo_to_der_round_trip()
    demo_oid_constants()
    demo_inspect_holder_and_issuer()
    print("\nAll AC examples completed.")


if __name__ == "__main__":
    main()