证书固定（Certificate Pinning）实战：原理、实现与绕过防御

class CertificatePinningInterceptor : Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response {
        val request = chain.request()
        val response = chain.proceed(request)
        
        // 获取 SSL 会话
        val sslSocket = response.chains.firstOrNull()?.socket as? SSLSocket
            ?: return response
            
        val session = sslSocket.session
        val peerCertificates = session.peerCertificates
        
        if (peerCertificates.isNotEmpty()) {
            val cert = peerCertificates[0] as? X509Certificate
                ?: throw IOException("Invalid certificate type")
            
            // 计算公钥 SHA-256 哈希
            val publicKey = cert.publicKey
            val publicKeyBytes = publicKey.encoded
            val sha256 = MessageDigest.getInstance("SHA-256")
                .digest(publicKeyBytes)
            val hash = Base64.encodeToString(sha256, Base64.NO_WRAP)
            
            // 与预置列表比较
            if (!ALLOWED_HASHES.contains(hash)) {
                throw IOException("Certificate pinning failure! " +
                    "Expected: ${ALLOWED_HASHES.joinToString()}, " +
                    "Got: $hash")
            }
        }
        
        return response
    }
}

private val ALLOWED_HASHES = setOf(
    "7HIpactkIAq2Y49orFOOQKurkmxEEuwFA9dNdJhSsEo=",
    "WdldG3m8L8XvNfVqK1nDZ3fFp7z5L0gE2y9XhKjTQ5c="
)

import TrustKit

func configureTrustKit() {
    let configuration: [String: Any] = [
        // 切换模式持续时间（秒），用于灰度发布
        kTSKSwitchingPolicyDuration: 86400,
        
        // 域名配置
        kTSKDomenPins: [
            "api.example.com": [
                // 固定公钥哈希列表
                kTSKPinnedPubKeys: [
                    "7HIpactkIAq2Y49orFOOQKurkmxEEuwFA9dNdJhSsEo=",
                    "WdldG3m8L8XvNfVqK1nDZ3fFp7z5L0gE2y9XhKjTQ5c="
                ],
                // 是否包含子域名
                kTSKIncludeSubdomains: true,
                // 是否强制验证（true=阻断，false=仅报告）
                kTSKEnforcement: true,
                // 报告端点（用于收集违规数据）
                kTSKReportUris: ["https://report.example.com/pinning-failures"]
            ]
        ]
    ]
    TrustKit.initWithoutConfiguration(configuration)
}

import ssl
import hashlib
import base64
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import serialization

def get_public_key_hash(domain: str, port: int = 443) -> str:
    """
    获取服务端证书公钥的 SHA-256 哈希（Base64 编码）
    """
    context = ssl.create_default_context()
    
    with socket.create_connection((domain, port)) as sock:
        with context.wrap_socket(sock, server_hostname=domain) as ssock:
            # 获取服务端证书
            cert_der = ssock.getpeercert(binary_form=True)
            
            # 解析证书并获取公钥
            cert = x509.load_der_x509_certificate(cert_der)
            public_key = cert.public_key()
            
            # 计算公钥哈希
            pubkey_bytes = public_key.public_bytes(
                encoding=serialization.Encoding.DER,
                format=serialization.PublicFormat.SubjectPublicKeyInfo
            )
            sha256_hash = hashlib.sha256(pubkey_bytes).digest()
            
            return base64.b64encode(sha256_hash).decode()

# 使用示例
if __name__ == "__main__":
    import socket
    domain = "api.example.com"
    print(f"Public key hash for {domain}:")
    print(get_public_key_hash(domain))

// frida script: bypass-pinning.js
Java.perform(function() {
    // 替换 OkHttp 的证书固定逻辑
    var CertificatePinner = Java.use("okhttp3.CertificatePinner");
    CertificatePinner.check.overload("java.lang.String", "java.util.List").implementation = function(host, peerCertificates) {
        console.log("[+] Certificate pinning bypassed for: " + host);
        return; // 直接返回，不校验
    };
    
    // 替换 TrustManager 验证
    var X509TrustManager = Java.use("javax.net.ssl.X509TrustManager");
    var TrustManager = Java.registerClass({
        name: "com.bypass.TrustManager",
        implements: [X509TrustManager],
        methods: {
            checkServerTrusted: function(chain, authType) {},
            checkClientTrusted: function(chain, authType) {},
            getAcceptedIssuers: function() { return []; }
        }
    });
    
    var SSLContext = Java.use("javax.net.ssl.SSLContext");
    SSLContext.init.overload("[Ljavax.net.ssl.KeyManager;", "[Ljavax.net.ssl.TrustManager;", "java.security.SecureRandom")
        .implementation = function(km, tm, sr) {
            this.init(km, [TrustManager.$new], sr);
        };
});

import ssl, hashlib, base64, socket
from cryptography import x509
from cryptography.hazmat.primitives import serialization

def get_pubkey_pin(domain, port=443):
    context = ssl.create_default_context()
    with socket.create_connection((domain, port)) as sock:
        with context.wrap_socket(sock, server_hostname=domain) as ssock:
            cert_der = ssock.getpeercert(binary_form=True)
            cert = x509.load_der_x509_certificate(cert_der)
            pubkey = cert.public_key()
            pubkey_bytes = pubkey.public_bytes(
                encoding=serialization.Encoding.DER,
                format=serialization.PublicFormat.SubjectPublicKeyInfo
            )
            digest = hashlib.sha256(pubkey_bytes).digest()
            return base64.b64encode(digest).decode()

print(f"Pin: {get_pubkey_pin('api.example.com')}")