如何验证ADFS加密的SAML响应(断言)

基础概念

Active Directory Federation Services (ADFS) 是微软提供的一种身份联合服务，用于在不同安全域之间实现单点登录（SSO）。SAML（Security Assertion Markup Language）是一种基于XML的标准，用于在不同的安全域之间交换身份验证和授权数据。

验证ADFS加密的SAML响应（断言）

验证ADFS加密的SAML响应涉及以下几个步骤：

解密SAML断言：首先需要解密由ADFS加密的SAML断言。
验证签名：确保SAML断言未被篡改。
验证断言内容：检查断言中的信息是否有效和可信。

解密SAML断言

ADFS使用对称加密算法（如AES）来加密SAML断言。解密过程需要使用ADFS配置中的密钥。

from lxml import etree
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import base64

def decrypt_saml_assertion(encrypted_assertion, key):
    # 解码Base64加密的断言
    encrypted_data = base64.b64decode(encrypted_assertion)
    
    # 解析加密数据
    iv = encrypted_data[:16]
    ciphertext = encrypted_data[16:]
    
    # 创建解密器
    cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
    decryptor = cipher.decryptor()
    
    # 解密数据
    decrypted_data = decryptor.update(ciphertext) + decryptor.finalize()
    
    # 解析XML
    return etree.fromstring(decrypted_data)

验证签名

ADFS使用非对称加密算法（如RSA）来签名SAML断言。验证签名需要使用ADFS的公钥。

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

def verify_saml_signature(saml_assertion, adfs_public_key):
    # 解析SAML断言
    root = etree.fromstring(saml_assertion)
    
    # 获取签名节点
    signature_node = root.find('.//{http://www.w3.org/2000/09/xmldsig#}Signature')
    
    # 获取签名值和签名数据
    signed_info = signature_node.find('.//{http://www.w3.org/2000/09/xmldsig#}SignedInfo')
    signature_value = signature_node.find('.//{http://www.w3.org/2000/09/xmldsig#}SignatureValue').text
    
    # 序列化公钥
    public_key = serialization.load_pem_public_key(adfs_public_key, backend=default_backend())
    
    # 验证签名
    public_key.verify(
        base64.b64decode(signature_value),
        etree.tostring(signed_info),
        padding.PKCS1v15(),
        hashes.SHA256()
    )

验证断言内容

验证断言内容包括检查断言的发行者、受众、有效期等信息。

def validate_saml_assertion(assertion, issuer, audience):
    # 检查发行者
    if assertion.issuer.text != issuer:
        raise ValueError("Invalid issuer")
    
    # 检查受众
    if audience not in assertion.conditions.attribute_statement:
        raise ValueError("Invalid audience")
    
    # 检查有效期
    if not assertion.conditions.not_on_or_after > datetime.utcnow():
        raise ValueError("Assertion has expired")