使用Python订阅Microsoft Graph在线会议转录文件变更通知时RSA解密失败的问题求助
使用Python订阅Microsoft Graph在线会议转录文件变更通知时RSA解密失败的问题求助
我现在正在尝试用Python订阅Microsoft Graph的communications/onlineMeetings/getAllTranscripts通知,并用自签名X.509证书解密收到的 payload。目前订阅和接收通知都能成功,但用RSA私钥解密AES密钥的操作一直失败。
这是我用来创建X.509证书的代码:
from cryptography import x509 from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives import hashes from datetime import datetime, timedelta, timezone def create_x509_certificate() -> tuple[x509.Certificate, rsa.RSAPrivateKey]: private_key = rsa.generate_private_key(public_exponent=65537, key_size=4096) subject = issuer = x509.Name( [ x509.NameAttribute(NameOID.COMMON_NAME, "Self-Signed Certificate"), ] ) cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(issuer) .public_key(private_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.now(timezone.utc)) .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365)) .sign(private_key, hashes.SHA256()) ) return (cert, private_key)
下面是我用来订阅通知的代码,这里我把证书编码成DER格式:
import base64 import time import requests client_state = APP.config.setdefault("client_state", str(int(time.time()))) expiration_time = (datetime.now(timezone.utc) + timedelta(minutes=120)).isoformat() cert, private_key = create_x509_certificate() APP.config["private_key"] = private_key b64_encoded_der_public_key = base64.b64encode(cert.public_bytes(Encoding.DER)).decode() LOGGER.debug("Public key: '%s'", b64_encoded_der_public_key) subscription_payload = { "resource": "communications/onlineMeetings/getAllTranscripts", "changeType": "created", "notificationUrl": f"{SERVER_URL}/webhook", "lifecycleNotificationUrl": f"{SERVER_URL}/webhook", "expirationDateTime": expiration_time, "clientState": client_state, "includeResourceData": True, "encryptionCertificate": b64_encoded_der_public_key, "encryptionCertificateId": f"CertificateId-{int(time.time())}", } response = requests.post( "https://graph.microsoft.com/v1.0/subscriptions", headers=_get_auth_headers(), json=subscription_payload, timeout=(5, 60), )
收到加密payload后,我尝试用下面的函数解密:
from cryptography.hazmat.primitives.asymmetric.padding import OAEP, MGF1 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives import hashes, hmac from cryptography.hazmat.primitives.padding import PKCS7 import base64 def _decrypt_payload( private_key: rsa.RSAPrivateKey, b64_encoded_payload: str, b64_encoded_data_key: str, b64_encoded_data_signature: str, ) -> str: # Decrypt the symmetric key symmetric_key = private_key.decrypt( base64.b64decode(b64_encoded_data_key), OAEP( mgf=MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None, ), ) encrypted_payload = base64.b64decode(b64_encoded_payload) # Create HMAC using decrypted symmetric key h = hmac.HMAC(symmetric_key, hashes.SHA256()) h.update(encrypted_payload) calculated_signature = base64.b64encode(h.finalize()).decode() # Verify signature if b64_encoded_data_signature != calculated_signature: raise Exception("Signature is invalid. Possible tampering detected.") # Decrypt the content iv = symmetric_key[:16] # First 16 bytes of the symmetric key as IV cipher = Cipher(algorithms.AES(symmetric_key), modes.CBC(iv)) decryptor = cipher.decryptor() decrypted_payload = decryptor.update(encrypted_payload) + decryptor.finalize() unpadder = PKCS7(algorithms.AES.block_size).unpadder() unpadded_data = unpadder.update(decrypted_payload) + unpadder.finalize() return unpadded_data.decode()
问题:
解密AES密钥时,private_key.decrypt()操作一直失败。我怀疑问题可能出在订阅时证书的编码方式,或者解密payload时的处理逻辑上。
疑问:
- 我在订阅请求的
encryptionCertificate字段中,把证书转成DER格式再base64编码的方式是正确的吗? - 我解密对称密钥或者验证HMAC签名的方式有没有问题?
- RSA密钥长度(4096位)或者AES密钥的提取逻辑会不会导致问题?
任何建议或见解都非常感谢!
备注:内容来源于stack exchange,提问作者Jims
相关产品推荐
相关产品推荐

