ASN.1 与 Python：使用 asn1crypto 处理网络协议数据

pip install asn1crypto

pip install --upgrade pip

python -c "import asn1crypto; print(asn1crypto.__version__)"

MySequence ::= SEQUENCE {
    myInt    INTEGER,
    myString UTF8String
}

30 0C 02 01 05 0C 05 48 65 6C 6C 6F

from asn1crypto.core import Sequence, Integer, UTF8String

# 定义 ASN.1 结构
class MySequence(Sequence):
    _fields = [
        ('myInt', Integer),
        ('myString', UTF8String),
    ]

# DER 编码数据（字节串）
der_data = b'\x30\x0C\x02\x01\x05\x0C\x05\x48\x65\x6C\x6C\x6F'

# 解析数据
my_sequence = MySequence.load(der_data)

# 访问数据
print(f"myInt: {my_sequence['myInt'].native}")  # 输出: myInt: 5
print(f"myString: {my_sequence['myString'].native}")  # 输出: myString: Hello

from asn1crypto import pem
from asn1crypto.x509 import Certificate

# 读取证书文件
with open('certificate.crt', 'rb') as f:
    certificate_data = f.read()

# 检查是否为 PEM 格式
if pem.detect(certificate_data):
    # 解码 PEM 格式
    type_name, headers, certificate_data = pem.unarmor(certificate_data)
    print(f"PEM 类型: {type_name}")

# 解析证书
cert = Certificate.load(certificate_data)

# 访问证书信息
print(f"版本: {cert['tbs_certificate']['version'].native}")
print(f"序列号: {cert['tbs_certificate']['serial_number'].native}")
print(f"颁发者: {cert['tbs_certificate']['issuer'].rfc4514_string()}")
print(f"主题: {cert['tbs_certificate']['subject'].rfc4514_string()}")
print(f"有效期开始: {cert['tbs_certificate']['validity']['not_before'].native}")
print(f"有效期结束: {cert['tbs_certificate']['validity']['not_after'].native}")

# 访问公钥信息
public_key = cert.public_key
print(f"公钥算法: {public_key.algorithm}")
print(f"公钥位数: {public_key.bit_size}")

from asn1crypto import pem
from asn1crypto.pkcs12 import PFX
from asn1crypto.x509 import Certificate
from asn1crypto.keys import PrivateKeyInfo

# 读取 PKCS #12 文件
with open('certificate.pfx', 'rb') as f:
    pfx_data = f.read()

# 密码（如果 PKCS #12 文件有密码保护）
password = 'your_password'

# 解析 PKCS #12 文件
pfx = PFX.load(pfx_data)
auth_safe = pfx['auth_safe']

# 检查是否有加密内容
if auth_safe.name == 'encrypted_data':
    # 解密加密内容
    decrypted = auth_safe.decrypt(password.encode('utf-8'))
    auth_safe = auth_safe['content'].load(decrypted)

# 提取证书和私钥
certificates = []
private_key = None

for content_info in auth_safe['content']:
    if content_info['content_type'].native == 'certificate':
        # 提取证书
        cert = content_info['content']['certificate']
        if cert.name == 'certificate':
            certificates.append(cert['tbs_certificate'])
    elif content_info['content_type'].native == 'key_bag':
        # 提取私钥
        private_key_info = content_info['content']['encrypted_data'].decrypt(
            password.encode('utf-8')
        )
        private_key = PrivateKeyInfo.load(private_key_info)

# 打印证书和私钥信息
print(f"找到 {len(certificates)} 个证书")
for i, cert in enumerate(certificates):
    print(f"证书 {i+1} 主题: {cert['subject'].rfc4514_string()}")

if private_key:
    print(f"私钥算法: {private_key.algorithm}")
    print(f"私钥位数: {private_key.bit_size}")
else:
    print("未找到私钥")

from asn1crypto.core import Sequence, Integer, UTF8String

# 定义 ASN.1 结构
class MySequence(Sequence):
    _fields = [
        ('myInt', Integer),
        ('myString', UTF8String),
    ]

# 创建 Python 对象
my_data = {
    'myInt': 42,
    'myString': 'Hello, World!'
}

# 实例化 ASN.1 对象
my_sequence = MySequence(my_data)

# 编码为 DER 格式
der_data = my_sequence.dump()

# 打印 DER 编码数据（十六进制表示）
print(f"DER 编码数据: {der_data.hex()}")

# 验证：重新解析 DER 数据
parsed_sequence = MySequence.load(der_data)
print(f"解析后的 myInt: {parsed_sequence['myInt'].native}")
print(f"解析后的 myString: {parsed_sequence['myString'].native}")

from asn1crypto import pem
from asn1crypto.csr import CertificationRequest, CertificationRequestInfo
from asn1crypto.algos import SignedDigestAlgorithm
from asn1crypto.x509 import Name, RelativeDistinguishedName, AttributeTypeAndValue
from asn1crypto.keys import PrivateKeyInfo, PublicKeyInfo
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes, serialization

# 生成 RSA 私钥
private_key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048,
    backend=default_backend()
)

# 导出私钥为 PEM 格式
private_key_pem = private_key.private_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption()
)

# 从私钥获取公钥
public_key = private_key.public_key()
public_key_bytes = public_key.public_bytes(
    encoding=serialization.Encoding.DER,
    format=serialization.PublicFormat.SubjectPublicKeyInfo
)

# 解析公钥
public_key_info = PublicKeyInfo.load(public_key_bytes)

# 创建主题 DN
subject = Name.build({
    'country_name': 'CN',
    'state_or_province_name': 'Beijing',
    'locality_name': 'Beijing',
    'organization_name': 'Example Company',
    'common_name': 'example.com',
})

# 创建扩展（可选）
extensions = [
    {
        'extn_id': 'subjectAltName',
        'critical': False,
        'extn_value': {
            'general_names': [
                {'dns_name': 'example.com'},
                {'dns_name': 'www.example.com'}
            ]
        }
    }
]

# 创建证书请求信息
csr_info = CertificationRequestInfo({
    'version': 'v1',
    'subject': subject,
    'subject_public_key_info': public_key_info,
    'attributes': [
        {
            'type': 'extension_request',
            'values': [extensions]
        }
    ]
})

# 对证书请求信息进行签名
csr_info_bytes = csr_info.dump()
signature_algorithm = SignedDigestAlgorithm({
    'algorithm': 'sha256_rsa'
})

# 使用 cryptography 库进行签名
signer = private_key.signer(
    padding.PKCS1v15(),
    hashes.SHA256()
)
signer.update(csr_info_bytes)
signature = signer.finalize()

# 创建证书签名请求
csr = CertificationRequest({
    'certification_request_info': csr_info,
    'signature_algorithm': signature_algorithm,
    'signature': signature
})

# 编码为 DER 格式
der_csr = csr.dump()

# 转换为 PEM 格式
pem_csr = pem.armor('CERTIFICATE REQUEST', der_csr)

# 保存到文件
with open('csr.pem', 'wb') as f:
    f.write(pem_csr)

print("CSR 已生成并保存到 csr.pem")

import socket
import ssl
from asn1crypto import pem, x509
from asn1crypto.util import timezone
from datetime import datetime

def get_server_certificate(hostname, port=443):
    """获取服务器证书"""
    context = ssl.create_default_context()
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE

    with socket.create_connection((hostname, port)) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            der_cert = ssock.getpeercert(binary_form=True)

    return der_cert

def verify_certificate_chain(cert_der, trusted_certs=None):
    """验证证书链"""
    # 解析服务器证书
    server_cert = x509.Certificate.load(cert_der)

    # 检查证书有效期
    now = datetime.now(timezone.utc)
    if now < server_cert['tbs_certificate']['validity']['not_before'].native:
        print("证书尚未生效")
        return False
    if now > server_cert['tbs_certificate']['validity']['not_after'].native:
        print("证书已过期")
        return False

    # 获取证书链（简化版，实际应用中可能需要从服务器获取完整链）
    certificate_chain = [server_cert]

    # 检查证书链中的每个证书
    for i, cert in enumerate(certificate_chain):
        # 最后一个证书应该由信任的根 CA 签名
        if i == len(certificate_chain) - 1:
            if trusted_certs:
                trusted = False
                for trusted_cert in trusted_certs:
                    if cert.issuer == trusted_cert.subject and cert.verify(trusted_cert):
                        trusted = True
                        break
                if not trusted:
                    print("证书未由受信任的 CA 签名")
                    return False
            else:
                print("没有提供信任的 CA 证书")
                return False
        else:
            # 中间证书应该由链中的下一个证书签名
            issuer_cert = certificate_chain[i + 1]
            if not cert.verify(issuer_cert):
                print(f"证书链验证失败：证书 {i} 未由证书 {i+1} 签名")
                return False

    print("证书链验证成功")
    return True

# 获取服务器证书
hostname = 'www.google.com'
cert_der = get_server_certificate(hostname)

# 加载系统信任的 CA 证书（简化版，实际应用中可能需要从系统证书存储中加载）
trusted_certs = []
try:
    with open('/etc/ssl/certs/ca-certificates.crt', 'rb') as f:
        pem_data = f.read()
        for type_name, headers, der_bytes in pem.unarmor(pem_data, multiple=True):
            if type_name == 'CERTIFICATE':
                trusted_certs.append(x509.Certificate.load(der_bytes))
except Exception as e:
    print(f"无法加载系统信任的 CA 证书: {e}")
    trusted_certs = None

# 验证证书链
verify_certificate_chain(cert_der, trusted_certs)