use pyo3::prelude::*;
use pyo3::types::PyBytes;
use synta::tag::{TagClass, TAG_SEQUENCE};
use synta::traits::{Decode, Encode};
use synta::{Decoder, Encoder, Encoding};
use synta_krb5::constants::{
ETYPE_AES128_CTS_HMAC_SHA1_96, ETYPE_AES128_CTS_HMAC_SHA256_128, ETYPE_AES256_CTS_HMAC_SHA1_96,
ETYPE_AES256_CTS_HMAC_SHA384_192, ETYPE_CAMELLIA128_CTS_CMAC, ETYPE_CAMELLIA256_CTS_CMAC,
ETYPE_DES3_CBC_MD5, ETYPE_DES3_CBC_SHA1, ETYPE_DES3_CBC_SHA1_KD, ETYPE_DES_CBC_CRC,
ETYPE_DES_CBC_MD4, ETYPE_DES_CBC_MD5, ETYPE_DES_HMAC_SHA1, ETYPE_RC4_HMAC, ETYPE_RC4_HMAC_EXP,
ID_PKINIT_SAN_COMPONENTS, NT_ENTERPRISE, NT_PRINCIPAL, NT_SMTP_NAME, NT_SRV_HST,
NT_SRV_HST_DOMAIN, NT_SRV_INST, NT_SRV_XHST, NT_UID, NT_UNKNOWN, NT_WELLKNOWN,
NT_X500_PRINCIPAL,
};
use synta_krb5::kerberos_v5::Int32;
use crate::error::SyntaErr;
use crate::types::PyObjectIdentifier;
fn oid_const(py: Python<'_>, components: &[u32]) -> PyResult<Py<pyo3::PyAny>> {
use synta::ObjectIdentifier;
let inner = ObjectIdentifier::new(components).expect("hardcoded OID is always valid");
Ok(Py::new(py, PyObjectIdentifier::from_oid(inner))?.into_any())
}
fn write_sequence(enc: &mut Encoder, inner: &[u8]) -> Result<(), synta::Error> {
enc.write_tag(synta::Tag::universal_constructed(TAG_SEQUENCE))?;
enc.write_length(inner.len())?;
enc.write_bytes(inner);
Ok(())
}
fn write_explicit_ctx(enc: &mut Encoder, tag_num: u32, inner: &[u8]) -> Result<(), synta::Error> {
enc.write_tag(synta::Tag::new(TagClass::ContextSpecific, true, tag_num))?;
enc.write_length(inner.len())?;
enc.write_bytes(inner);
Ok(())
}
fn encode_to_vec<T: Encode>(value: &T) -> Result<Vec<u8>, synta::Error> {
let mut enc = Encoder::new(Encoding::Der);
value.encode(&mut enc)?;
enc.finish()
}
fn read_tag_check(
dec: &mut Decoder<'_>,
class: TagClass,
number: u32,
constructed: bool,
label: &'static str,
) -> PyResult<()> {
let tag = dec.read_tag().map_err(SyntaErr)?;
if tag.class() != class || tag.number() != number || tag.is_constructed() != constructed {
return Err(pyo3::exceptions::PyValueError::new_err(format!(
"KRB5PrincipalName: expected {label} (class={class:?}, tag={number}), \
got class={:?} tag={} constructed={}",
tag.class(),
tag.number(),
tag.is_constructed()
)));
}
Ok(())
}
fn read_definite_len(dec: &mut Decoder<'_>) -> PyResult<usize> {
Ok(dec
.read_length()
.map_err(SyntaErr)?
.definite()
.map_err(SyntaErr)?)
}
fn read_content<'a>(
dec: &mut Decoder<'a>,
class: TagClass,
number: u32,
label: &'static str,
) -> PyResult<&'a [u8]> {
read_tag_check(dec, class, number, true, label)?;
let len = read_definite_len(dec)?;
Ok(dec.read_bytes(len).map_err(SyntaErr)?)
}
fn read_sequence<'a>(dec: &mut Decoder<'a>, label: &'static str) -> PyResult<&'a [u8]> {
read_content(dec, TagClass::Universal, TAG_SEQUENCE, label)
}
fn read_explicit_ctx<'a>(
dec: &mut Decoder<'a>,
tag_num: u32,
label: &'static str,
) -> PyResult<&'a [u8]> {
read_content(dec, TagClass::ContextSpecific, tag_num, label)
}
fn encode_krb5principalname(
realm: &str,
name_type: i32,
components: &[String],
) -> PyResult<Vec<u8>> {
let ns_items: Result<Vec<Vec<u8>>, synta::Error> = components
.iter()
.map(|c| encode_to_vec(&synta::GeneralString::new(c.as_bytes().to_vec())))
.collect();
let ns_items = ns_items.map_err(SyntaErr)?;
let ns_flat: Vec<u8> = ns_items.concat();
let mut ns_seq_enc = Encoder::new(Encoding::Der);
write_sequence(&mut ns_seq_enc, &ns_flat).map_err(SyntaErr)?;
let ns_seq_bytes = ns_seq_enc.finish().map_err(SyntaErr)?;
let nt_bytes = encode_to_vec(&synta::Integer::from_i64(name_type as i64)).map_err(SyntaErr)?;
let mut nt_ctx = Encoder::new(Encoding::Der);
write_explicit_ctx(&mut nt_ctx, 0, &nt_bytes).map_err(SyntaErr)?;
let nt_ctx_bytes = nt_ctx.finish().map_err(SyntaErr)?;
let mut ns_ctx = Encoder::new(Encoding::Der);
write_explicit_ctx(&mut ns_ctx, 1, &ns_seq_bytes).map_err(SyntaErr)?;
let ns_ctx_bytes = ns_ctx.finish().map_err(SyntaErr)?;
let pn_inner = [nt_ctx_bytes, ns_ctx_bytes].concat();
let mut pn_enc = Encoder::new(Encoding::Der);
write_sequence(&mut pn_enc, &pn_inner).map_err(SyntaErr)?;
let pn_bytes = pn_enc.finish().map_err(SyntaErr)?;
let realm_gs_bytes =
encode_to_vec(&synta::GeneralString::new(realm.as_bytes().to_vec())).map_err(SyntaErr)?;
let mut realm_ctx = Encoder::new(Encoding::Der);
write_explicit_ctx(&mut realm_ctx, 0, &realm_gs_bytes).map_err(SyntaErr)?;
let realm_ctx_bytes = realm_ctx.finish().map_err(SyntaErr)?;
let mut pn_ctx = Encoder::new(Encoding::Der);
write_explicit_ctx(&mut pn_ctx, 1, &pn_bytes).map_err(SyntaErr)?;
let pn_ctx_bytes = pn_ctx.finish().map_err(SyntaErr)?;
let outer_content = [realm_ctx_bytes, pn_ctx_bytes].concat();
let mut outer = Encoder::new(Encoding::Der);
write_sequence(&mut outer, &outer_content).map_err(SyntaErr)?;
Ok(outer.finish().map_err(SyntaErr)?)
}
fn decode_krb5principalname(data: &[u8]) -> PyResult<(String, i32, Vec<String>)> {
let ve = |s: &str| pyo3::exceptions::PyValueError::new_err(s.to_string());
let mut outer = Decoder::new(data, Encoding::Der);
let outer_content = read_sequence(&mut outer, "outer SEQUENCE")?;
let mut seq = Decoder::new(outer_content, Encoding::Der);
let realm_bytes = read_explicit_ctx(&mut seq, 0, "[0] realm")?;
let mut realm_dec = Decoder::new(realm_bytes, Encoding::Der);
let realm_gs = synta::GeneralString::decode(&mut realm_dec).map_err(SyntaErr)?;
let realm = String::from_utf8(realm_gs.as_bytes().to_vec())
.map_err(|_| ve("KRB5PrincipalName: realm is not valid UTF-8"))?;
let pn_bytes = read_explicit_ctx(&mut seq, 1, "[1] principalName")?;
let mut pn_dec = Decoder::new(pn_bytes, Encoding::Der);
let pn_inner = read_sequence(&mut pn_dec, "PrincipalName SEQUENCE")?;
let mut pn_inner_dec = Decoder::new(pn_inner, Encoding::Der);
let nt_bytes = read_explicit_ctx(&mut pn_inner_dec, 0, "[0] name-type")?;
let mut nt_dec = Decoder::new(nt_bytes, Encoding::Der);
let int32 = Int32::decode(&mut nt_dec).map_err(SyntaErr)?;
let name_type = int32.get();
let ns_ctx_bytes = read_explicit_ctx(&mut pn_inner_dec, 1, "[1] name-string")?;
let mut ns_ctx_dec = Decoder::new(ns_ctx_bytes, Encoding::Der);
let ns_bytes = read_sequence(&mut ns_ctx_dec, "name-string SEQUENCE")?;
let mut components = Vec::new();
let mut ns_dec = Decoder::new(ns_bytes, Encoding::Der);
while !ns_dec.is_empty() {
let gs = synta::GeneralString::decode(&mut ns_dec).map_err(SyntaErr)?;
let s = String::from_utf8(gs.as_bytes().to_vec())
.map_err(|_| ve("KRB5PrincipalName: name component is not valid UTF-8"))?;
components.push(s);
}
Ok((realm, name_type, components))
}
#[pyclass(name = "Krb5PrincipalName", frozen)]
pub struct PyKrb5PrincipalName {
realm: String,
name_type: i32,
components: Vec<String>,
}
#[pymethods]
impl PyKrb5PrincipalName {
#[new]
#[pyo3(signature = (realm, name_type, components))]
fn new(realm: String, name_type: i32, components: Vec<String>) -> PyResult<Self> {
if !realm.is_ascii() {
return Err(pyo3::exceptions::PyValueError::new_err(
"realm must be an ASCII string",
));
}
for comp in &components {
if !comp.is_ascii() {
return Err(pyo3::exceptions::PyValueError::new_err(format!(
"name component {comp:?} must be an ASCII string"
)));
}
}
Ok(Self {
realm,
name_type,
components,
})
}
#[getter]
fn realm(&self) -> &str {
&self.realm
}
#[getter]
fn name_type(&self) -> i32 {
self.name_type
}
#[getter]
fn components<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, pyo3::types::PyList>> {
pyo3::types::PyList::new(py, self.components.iter())
}
fn to_der<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
let der = encode_krb5principalname(&self.realm, self.name_type, &self.components)?;
Ok(PyBytes::new(py, &der))
}
#[staticmethod]
fn from_der(data: &[u8]) -> PyResult<Self> {
let (realm, name_type, components) = decode_krb5principalname(data)?;
Ok(Self {
realm,
name_type,
components,
})
}
fn __repr__(&self) -> String {
format!(
"Krb5PrincipalName(realm={:?}, name_type={}, components={:?})",
self.realm, self.name_type, self.components
)
}
fn __eq__(&self, other: PyRef<'_, Self>) -> bool {
self.realm == other.realm
&& self.name_type == other.name_type
&& self.components == other.components
}
}
pub fn register_krb5_module(parent: &Bound<'_, PyModule>) -> PyResult<()> {
let py = parent.py();
let m = PyModule::new(py, "krb5")?;
m.add("NT_UNKNOWN", NT_UNKNOWN)?;
m.add("NT_PRINCIPAL", NT_PRINCIPAL)?;
m.add("NT_SRV_INST", NT_SRV_INST)?;
m.add("NT_SRV_HST", NT_SRV_HST)?;
m.add("NT_SRV_XHST", NT_SRV_XHST)?;
m.add("NT_UID", NT_UID)?;
m.add("NT_X500_PRINCIPAL", NT_X500_PRINCIPAL)?;
m.add("NT_SMTP_NAME", NT_SMTP_NAME)?;
m.add("NT_ENTERPRISE", NT_ENTERPRISE)?;
m.add("NT_WELLKNOWN", NT_WELLKNOWN)?;
m.add("NT_SRV_HST_DOMAIN", NT_SRV_HST_DOMAIN)?;
m.add("ETYPE_DES_CBC_CRC", ETYPE_DES_CBC_CRC)?;
m.add("ETYPE_DES_CBC_MD4", ETYPE_DES_CBC_MD4)?;
m.add("ETYPE_DES_CBC_MD5", ETYPE_DES_CBC_MD5)?;
m.add("ETYPE_DES3_CBC_MD5", ETYPE_DES3_CBC_MD5)?;
m.add("ETYPE_DES3_CBC_SHA1", ETYPE_DES3_CBC_SHA1)?;
m.add("ETYPE_DES_HMAC_SHA1", ETYPE_DES_HMAC_SHA1)?;
m.add("ETYPE_DES3_CBC_SHA1_KD", ETYPE_DES3_CBC_SHA1_KD)?;
m.add(
"ETYPE_AES128_CTS_HMAC_SHA1_96",
ETYPE_AES128_CTS_HMAC_SHA1_96,
)?;
m.add(
"ETYPE_AES256_CTS_HMAC_SHA1_96",
ETYPE_AES256_CTS_HMAC_SHA1_96,
)?;
m.add(
"ETYPE_AES128_CTS_HMAC_SHA256_128",
ETYPE_AES128_CTS_HMAC_SHA256_128,
)?;
m.add(
"ETYPE_AES256_CTS_HMAC_SHA384_192",
ETYPE_AES256_CTS_HMAC_SHA384_192,
)?;
m.add("ETYPE_RC4_HMAC", ETYPE_RC4_HMAC)?;
m.add("ETYPE_RC4_HMAC_EXP", ETYPE_RC4_HMAC_EXP)?;
m.add("ETYPE_CAMELLIA128_CTS_CMAC", ETYPE_CAMELLIA128_CTS_CMAC)?;
m.add("ETYPE_CAMELLIA256_CTS_CMAC", ETYPE_CAMELLIA256_CTS_CMAC)?;
m.add(
"KRB5_PRINCIPAL_NAME_OID",
oid_const(py, ID_PKINIT_SAN_COMPONENTS)?,
)?;
m.add_class::<PyKrb5PrincipalName>()?;
crate::pkinit::register_pkinit_classes(&m)?;
crate::install_submodule(
parent,
&m,
"synta.krb5",
Some(
"Kerberos V5 ASN.1 types: principal names, PKINIT protocol types, \
name-type constants (NT_*), and encryption-type constants (ETYPE_*).",
),
)
}