#!/usr/bin/env python3 import base64 import hashlib import os import re import subprocess import sys import tempfile from datetime import datetime, timezone from pathlib import Path from asn1crypto import cms, core, pem, x509 OPENSSL_CONF = "" CERT_URI = "" KEY_URI = "" CHAIN_PEM = "" def load_config(config_path: str) -> None: global OPENSSL_CONF, CERT_URI, KEY_URI, CHAIN_PEM OPENSSL_CONF = config_path eml_conf = read_simple_section(OPENSSL_CONF, "eml_sign") CERT_URI = eml_conf["cert_uri"] KEY_URI = eml_conf["key_uri"] CHAIN_PEM = eml_conf["chain_pem"] def read_simple_section(path: str, section: str) -> dict[str, str]: result = {} active = False with open(path, "r", encoding="utf-8") as f: for raw_line in f: line = raw_line.strip() if not line: continue if line.startswith("#") or line.startswith(";"): continue if line.startswith("[") and line.endswith("]"): active = (line[1:-1].strip() == section) continue if not active: continue if "=" not in line: continue key, value = line.split("=", 1) result[key.strip()] = value.strip() return result def to_crlf(data: bytes) -> bytes: data = data.replace(b"\r\n", b"\n").replace(b"\r", b"\n") return data.replace(b"\n", b"\r\n") def split_headers_body(raw: bytes) -> tuple[bytes, bytes]: if b"\r\n\r\n" in raw: return raw.split(b"\r\n\r\n", 1) if b"\n\n" in raw: return raw.split(b"\n\n", 1) return b"", raw def split_header_lines(header_block: bytes) -> list[bytes]: lines = to_crlf(header_block).split(b"\r\n") result = [] current = b"" for line in lines: if not line: continue if line.startswith((b" ", b"\t")) and current: current += b"\r\n" + line else: if current: result.append(current) current = line if current: result.append(current) return result def header_name(header: bytes) -> str: return header.split(b":", 1)[0].decode("ascii", errors="ignore").lower() def load_cert_from_pkcs11_uri(cert_uri: str) -> x509.Certificate: env = os.environ.copy() env["OPENSSL_CONF"] = OPENSSL_CONF cmd = [ "openssl", "x509", "-provider", "pkcs11", "-provider", "default", "-in", cert_uri, "-outform", "DER", ] der = subprocess.check_output(cmd, env=env) return x509.Certificate.load(der) def load_chain_pem(path: str) -> list[x509.Certificate]: p = Path(path) if not p.is_file(): return [] data = p.read_bytes() matches = re.findall( rb"-----BEGIN CERTIFICATE-----\s*(.*?)\s*-----END CERTIFICATE-----", data, flags=re.DOTALL, ) certs = [] for body in matches: der = base64.b64decode(re.sub(rb"\s+", b"", body)) certs.append(x509.Certificate.load(der)) return certs def sign_with_token(data: bytes) -> bytes: env = os.environ.copy() env["OPENSSL_CONF"] = OPENSSL_CONF with tempfile.TemporaryDirectory() as tmpdir: data_file = Path(tmpdir) / "data-to-sign.der" sig_file = Path(tmpdir) / "signature.bin" data_file.write_bytes(data) cmd = [ "openssl", "dgst", "-sha256", "-provider", "pkcs11", "-provider", "default", "-sign", KEY_URI, "-out", str(sig_file), str(data_file), ] subprocess.check_call(cmd, env=env) return sig_file.read_bytes() def b64_lines(data: bytes, line_len: int = 64) -> str: txt = base64.b64encode(data).decode("ascii") return "\r\n".join( txt[i:i + line_len] for i in range(0, len(txt), line_len) ) def make_signed_eml(input_eml: bytes) -> bytes: header_block, body = split_headers_body(input_eml) headers = split_header_lines(header_block) outer_headers = [] inner_headers = [] for h in headers: name = header_name(h) if name.startswith("content-"): inner_headers.append(h) elif name == "mime-version": pass else: outer_headers.append(h) if not inner_headers: inner_headers.append(b"Content-Type: text/plain; charset=utf-8") inner_headers.append(b"Content-Transfer-Encoding: 8bit") inner_entity = b"\r\n".join(inner_headers) + b"\r\n\r\n" + to_crlf(body) signer_cert = load_cert_from_pkcs11_uri(CERT_URI) chain_certs = load_chain_pem(CHAIN_PEM) digest = hashlib.sha256(inner_entity).digest() signed_attrs = cms.CMSAttributes([ cms.CMSAttribute({ "type": "content_type", "values": ["data"], }), cms.CMSAttribute({ "type": "message_digest", "values": [digest], }), cms.CMSAttribute({ "type": "signing_time", "values": [ cms.Time({ "utc_time": core.UTCTime(datetime.now(timezone.utc)) }) ], }), ]) signature = sign_with_token(signed_attrs.dump()) signer_info = cms.SignerInfo({ "version": "v1", "sid": cms.SignerIdentifier({ "issuer_and_serial_number": cms.IssuerAndSerialNumber({ "issuer": signer_cert.issuer, "serial_number": signer_cert.serial_number, }) }), "digest_algorithm": cms.DigestAlgorithm({ "algorithm": "sha256", }), "signed_attrs": signed_attrs, "signature_algorithm": cms.SignedDigestAlgorithm({ "algorithm": "rsassa_pkcs1v15", }), "signature": signature, }) cms_certs = [ cms.CertificateChoices({ "certificate": signer_cert, }) ] for chain_cert in chain_certs: cms_certs.append( cms.CertificateChoices({ "certificate": chain_cert, }) ) signed_data = cms.SignedData({ "version": "v1", "digest_algorithms": [ cms.DigestAlgorithm({ "algorithm": "sha256", }) ], "encap_content_info": cms.ContentInfo({ "content_type": "data", "content": inner_entity, }), "certificates": cms_certs, "signer_infos": [ signer_info, ], }) content_info = cms.ContentInfo({ "content_type": "signed_data", "content": signed_data, }) cms_der = content_info.dump() new_headers = [] new_headers.extend(outer_headers) new_headers.append(b"MIME-Version: 1.0") new_headers.append( b'Content-Type: application/pkcs7-mime; smime-type=signed-data; name="smime.p7m"' ) new_headers.append(b"Content-Transfer-Encoding: base64") new_headers.append(b'Content-Disposition: attachment; filename="smime.p7m"') out = b"\r\n".join(new_headers) out += b"\r\n\r\n" out += b64_lines(cms_der).encode("ascii") out += b"\r\n" return out def main() -> int: if len(sys.argv) != 4: print( f"Použití: {sys.argv[0]} konfigurak.cnf vstup.eml vystup-signed.eml", file=sys.stderr, ) return 2 config_path = sys.argv[1] input_path = Path(sys.argv[2]) output_path = Path(sys.argv[3]) load_config(config_path) signed = make_signed_eml(input_path.read_bytes()) output_path.write_bytes(signed) return 0 if __name__ == "__main__": raise SystemExit(main())