use pyo3::prelude::*;
use pyo3::types::PyBytes;
use synta::traits::{Decode, Encode};
use synta::{Decoder, Encoding};
use crate::error::SyntaErr;
use crate::types::PyObjectIdentifier;
fn encode_to_der<T: Encode>(v: &T) -> Vec<u8> {
let mut enc = synta::Encoder::new(Encoding::Der);
if v.encode(&mut enc).is_err() {
return Vec::new();
}
enc.finish().unwrap_or_default()
}
#[pyclass(frozen, name = "MSCSTemplateV2")]
pub struct PyMSCSTemplateV2 {
template_id: Py<PyObjectIdentifier>,
template_major_version: i64,
template_minor_version: Option<i64>,
raw_der: Vec<u8>,
}
#[pymethods]
impl PyMSCSTemplateV2 {
#[staticmethod]
pub fn from_der(py: Python<'_>, data: &[u8]) -> PyResult<Self> {
let mut dec = Decoder::new(data, Encoding::Der);
let tmpl =
synta_certificate::ms_pki_types::MSCSTemplateV2::decode(&mut dec).map_err(SyntaErr)?;
let template_id = Py::new(py, PyObjectIdentifier::from_oid(tmpl.template_id.clone()))?;
let template_major_version = tmpl.template_major_version.as_i64().map_err(SyntaErr)?;
let template_minor_version = tmpl
.template_minor_version
.as_ref()
.map(|v| v.as_i64().map_err(SyntaErr))
.transpose()?;
let raw_der = encode_to_der(&tmpl);
Ok(PyMSCSTemplateV2 {
template_id,
template_major_version,
template_minor_version,
raw_der,
})
}
pub fn to_der<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
PyBytes::new(py, &self.raw_der)
}
#[getter]
fn template_id(&self, py: Python<'_>) -> Py<PyObjectIdentifier> {
self.template_id.clone_ref(py)
}
#[getter]
fn template_major_version(&self) -> i64 {
self.template_major_version
}
#[getter]
fn template_minor_version(&self) -> Option<i64> {
self.template_minor_version
}
fn __repr__(&self, py: Python<'_>) -> String {
let oid = self.template_id.bind(py).get();
format!(
"MSCSTemplateV2(template_id='{}', major={}, minor={:?})",
oid.inner, self.template_major_version, self.template_minor_version,
)
}
}
#[pyclass(frozen, name = "RequestClientInfo")]
pub struct PyRequestClientInfo {
client_id: Option<i64>,
machine_name: Option<String>,
user_name: Option<String>,
process_name: Option<String>,
}
#[pymethods]
impl PyRequestClientInfo {
#[staticmethod]
pub fn from_der(data: &[u8]) -> PyResult<Self> {
let mut dec = Decoder::new(data, Encoding::Der);
let info = synta_certificate::ms_pki_types::RequestClientInfo::decode(&mut dec)
.map_err(SyntaErr)?;
let client_id = info
.client_id
.as_ref()
.map(|v| v.as_i64().map_err(SyntaErr))
.transpose()?;
let machine_name = info.machine_name.as_ref().map(|s| s.as_str().to_owned());
let user_name = info.user_name.as_ref().map(|s| s.as_str().to_owned());
let process_name = info.process_name.as_ref().map(|s| s.as_str().to_owned());
Ok(PyRequestClientInfo {
client_id,
machine_name,
user_name,
process_name,
})
}
#[getter]
fn client_id(&self) -> Option<i64> {
self.client_id
}
#[getter]
fn machine_name(&self) -> Option<&str> {
self.machine_name.as_deref()
}
#[getter]
fn user_name(&self) -> Option<&str> {
self.user_name.as_deref()
}
#[getter]
fn process_name(&self) -> Option<&str> {
self.process_name.as_deref()
}
fn __repr__(&self) -> String {
format!(
"RequestClientInfo(machine={:?}, user={:?}, process={:?})",
self.machine_name, self.user_name, self.process_name,
)
}
}
pub(super) fn register_ms_pki_submodule(parent: &Bound<'_, PyModule>) -> PyResult<()> {
let py = parent.py();
let m = PyModule::new(py, "synta.ms_pki")?;
m.add_class::<PyMSCSTemplateV2>()?;
m.add_class::<PyRequestClientInfo>()?;
let oc = |components: &[u32]| super::oid_const(py, components);
m.add(
"ID_MS_CERTSRV_CA_VERSION",
oc(synta_certificate::ms_pki_types::ID_MS_CERTSRV_CA_VERSION),
)?;
m.add(
"ID_MS_CERTSRV_PREVIOUS_CERT_HASH",
oc(synta_certificate::ms_pki_types::ID_MS_CERTSRV_PREVIOUS_CERT_HASH),
)?;
m.add(
"ID_MS_CRL_VIRTUAL_BASE",
oc(synta_certificate::ms_pki_types::ID_MS_CRL_VIRTUAL_BASE),
)?;
m.add(
"ID_MS_CRL_NEXT_PUBLISH",
oc(synta_certificate::ms_pki_types::ID_MS_CRL_NEXT_PUBLISH),
)?;
m.add(
"ID_MS_KP_CA_EXCHANGE",
oc(synta_certificate::ms_pki_types::ID_MS_KP_CA_EXCHANGE),
)?;
m.add(
"ID_MS_KP_KEY_RECOVERY_AGENT",
oc(synta_certificate::ms_pki_types::ID_MS_KP_KEY_RECOVERY_AGENT),
)?;
m.add(
"ID_MS_ENTERPRISE_OID_ROOT",
oc(synta_certificate::ms_pki_types::ID_MS_ENTERPRISE_OID_ROOT),
)?;
m.add(
"ID_MS_APPLICATION_CERT_POLICIES",
oc(synta_certificate::ms_pki_types::ID_MS_APPLICATION_CERT_POLICIES),
)?;
m.add(
"ID_MS_REQUEST_CLIENT_INFO",
oc(synta_certificate::ms_pki_types::ID_MS_REQUEST_CLIENT_INFO),
)?;
m.add(
"ID_MS_ENCRYPTED_KEY_HASH",
oc(synta_certificate::ms_pki_types::ID_MS_ENCRYPTED_KEY_HASH),
)?;
m.add(
"ID_MS_CERTSRV_CROSSCA_VERSION",
oc(synta_certificate::ms_pki_types::ID_MS_CERTSRV_CROSSCA_VERSION),
)?;
m.add(
"ID_MS_KP_CTL_USAGE_SIGNING",
oc(synta_certificate::ms_pki_types::ID_MS_KP_CTL_USAGE_SIGNING),
)?;
m.add(
"ID_MS_KP_TIME_STAMP_SIGNING",
oc(synta_certificate::ms_pki_types::ID_MS_KP_TIME_STAMP_SIGNING),
)?;
m.add(
"ID_MS_KP_EFS_CRYPTO",
oc(synta_certificate::ms_pki_types::ID_MS_KP_EFS_CRYPTO),
)?;
m.add(
"ID_MS_KP_EFS_RECOVERY",
oc(synta_certificate::ms_pki_types::ID_MS_KP_EFS_RECOVERY),
)?;
m.add(
"ID_MS_KP_KEY_RECOVERY",
oc(synta_certificate::ms_pki_types::ID_MS_KP_KEY_RECOVERY),
)?;
m.add(
"ID_MS_KP_DOCUMENT_SIGNING",
oc(synta_certificate::ms_pki_types::ID_MS_KP_DOCUMENT_SIGNING),
)?;
m.add(
"ID_MS_KP_LIFETIME_SIGNING",
oc(synta_certificate::ms_pki_types::ID_MS_KP_LIFETIME_SIGNING),
)?;
m.add(
"ID_MS_AUTO_ENROLL_CTL_USAGE",
oc(synta_certificate::ms_pki_types::ID_MS_AUTO_ENROLL_CTL_USAGE),
)?;
crate::install_submodule(
parent,
&m,
"synta.ms_pki",
Some(concat!(
"synta.ms_pki — Microsoft PKI (AD CS) certificate extension types.\n\n",
"Provides ``MSCSTemplateV2`` for parsing certificate template extensions\n",
"and ``RequestClientInfo`` for parsing enrollment client info extensions.\n",
)),
)?;
Ok(())
}