一句话总结：CDP 的 Security 和 Audits 域可以让你编程地管理浏览器安全策略——处理证书警告、监控混合内容、捕获 CSP 违规，以及控制权限授予。

目录

  1. 为什么用 CDP 管理安全策略
  2. 处理证书错误
  3. 监控混合内容
  4. 捕获 CSP 违规
  5. 管理权限
  6. 实战：自动化安全测试
  7. 常见踩坑与最佳实践

为什么用 CDP 管理安全策略

在自动化测试中，证书错误和混合内容经常导致测试中断。CDP 提供了编程管理方案：

功能 手动处理 CDP 方案
自签名证书 手动点击”高级→继续” ✅ Security.setIgnoreCertificateErrors
混合内容警告 手动允许加载 ✅ Security.enable 监控
CSP 违规 看 Console 面板 ✅ Audits 域捕获
权限授予 手动点击允许 ✅ Browser.setPermission
安全事件 ❌ 无自动化 ✅ 完整事件流

处理证书错误

忽略证书错误

最常用的功能——在测试环境中忽略自签名证书：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import asyncio
import websockets
import json

CDP_URL = "ws://127.0.0.1:9222/devtools/browser/..."
CMD_ID = [0]

async def cdp(ws, session_id, method, params=None):
    CMD_ID[0] += 1; cmd_id = CMD_ID[0]
    await ws.send(json.dumps({"sessionId": session_id, "id": cmd_id,
                               "method": method, "params": params or {}}))
    async for msg in ws:
        resp = json.loads(msg)
        if resp.get("id") == cmd_id:
            return resp.get("result", {})

async def connect_page(ws):
    targets = await cdp(ws, None, "Target.getTargets")
    target_id = targets["targetInfos"][0]["targetId"]
    session = await cdp(ws, None, "Target.attachToTarget", {
        "targetId": target_id, "flatten": True
    })
    return session["sessionId"]


async def ignore_certificate_errors(ws, session_id, ignore=True):
    """
    设置是否忽略证书错误
    在导航到使用自签名证书的 HTTPS 页面时非常有用
    """
    return await cdp(ws, session_id, "Security.setIgnoreCertificateErrors", {
        "ignore": ignore
    })


async def navigate_with_insecure_cert(ws, session_id, url):
    """导航到证书有问题的 HTTPS 页面"""
    # 先设置忽略证书错误
    await ignore_certificate_errors(ws, session_id, True)
    
    # 导航
    result = await cdp(ws, session_id, "Page.navigate", {"url": url})
    await asyncio.sleep(3)
    
    print(f"已忽略证书错误，导航到: {url}")
    return result

监控安全状态变化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
async def enable_security_monitoring(ws, session_id):
    """启用安全事件监控"""
    return await cdp(ws, session_id, "Security.enable")


async def monitor_security_state(ws, session_id, duration=30):
    """监控页面安全状态变化"""
    await enable_security_monitoring(ws, session_id)
    
    events = []
    start = asyncio.get_event_loop().time()
    
    while (asyncio.get_event_loop().time() - start) < duration:
        try:
            msg = await asyncio.wait_for(ws.__anext__(), timeout=1)
            data = json.loads(msg)
            method = data.get("method", "")
            
            if method == "Security.securityStateChanged":
                p = data["params"]
                state = {
                    "security_state": p.get("securityState", ""),
                    "scheme_cryptographic": p.get("schemeCryptographic", False),
                    "explanations": p.get("explanations", []),
                }
                events.append(state)
                print(f"[安全状态] {state['security_state']}")
                
                for exp in p.get("explanations", []):
                    print(f"  - {exp.get('summary', '')}")
            
            elif method == "Security.certificateError":
                p = data["params"]
                print(f"[证书错误] {p.get('errorType', '')}: {p.get('requestURL', '')}")
                events.append(("certificate_error", p))
                
                # 自动处理证书错误
                await cdp(ws, session_id, "Security.handleCertificateError", {
                    "eventId": p.get("eventId"),
                    "action": "continue"  # 或 "deny"
                })
                    
        except asyncio.TimeoutError:
            continue
    
    return events

监控混合内容

混合内容（Mixed Content）指 HTTPS 页面加载 HTTP 资源。CDP 可以监控这些警告：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
async def monitor_mixed_content(ws, session_id, duration=30):
    """监控混合内容事件"""
    # 启用网络和安全域
    await cdp(ws, session_id, "Network.enable")
    await enable_security_monitoring(ws, session_id)
    
    mixed = []
    start = asyncio.get_event_loop().time()
    
    while (asyncio.get_event_loop().time() - start) < duration:
        try:
            msg = await asyncio.wait_for(ws.__anext__(), timeout=1)
            data = json.loads(msg)
            method = data.get("method", "")
            
            # 混合内容在安全状态变化中体现
            if method == "Security.securityStateChanged":
                for exp in data["params"].get("explanations", []):
                    if "mixed" in exp.get("summary", "").lower():
                        mixed.append({
                            "summary": exp.get("summary"),
                            "description": exp.get("description", ""),
                            "url": exp.get("url", "")
                        })
                        print(f"[混合内容] {exp.get('summary', '')}")
            
            # 网络加载失败也可能与混合内容相关
            elif method == "Network.loadingFailed":
                p = data["params"]
                if p.get("blockedReason") == "mixed-content":
                    mixed.append({
                        "type": "blocked",
                        "url": p.get("url", ""),
                        "error": p.get("errorText", "")
                    })
                    print(f"[混合内容被拦截] {p.get('url', '')}")
                    
        except asyncio.TimeoutError:
            continue
    
    return mixed

捕获 CSP 违规

Content Security Policy（CSP）违规可以通过 Audits 域或 Security 域捕获：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
async def enable_csp_monitoring(ws, session_id):
    """启用 CSP 违规监控"""
    # 通过 Security 域启用
    await enable_security_monitoring(ws, session_id)
    
    # 同时启用 Console 域（CSP 违规也会出现在 Console）
    await cdp(ws, session_id, "Console.enable")


async def monitor_csp_violations(ws, session_id, duration=30):
    """监控 CSP 违规"""
    violations = []
    start = asyncio.get_event_loop().time()
    
    while (asyncio.get_event_loop().time() - start) < duration:
        try:
            msg = await asyncio.wait_for(ws.__anext__(), timeout=1)
            data = json.loads(msg)
            method = data.get("method", "")
            
            # 从 Console 消息中捕获 CSP 违规
            if method == "Console.messageAdded":
                msg_data = data["params"]["message"]
                text = msg_data.get("text", "")
                if "CSP" in text or "Content Security Policy" in text:
                    violations.append({
                        "text": text,
                        "url": msg_data.get("url", ""),
                        "line": msg_data.get("line", 0),
                    })
                    print(f"[CSP 违规] {text[:100]}")
            
            # 从网络请求中捕获被 CSP 阻止的加载
            elif method == "Network.loadingFailed":
                p = data["params"]
                if p.get("blockedReason") in ("csp", "content-type"):
                    violations.append({
                        "type": "blocked_by_csp",
                        "url": p.get("url", ""),
                        "reason": p.get("blockedReason"),
                    })
                    print(f"[CSP 阻止] {p.get('url', '')}")
                    
        except asyncio.TimeoutError:
            continue
    
    return violations

管理权限

编程授予/拒绝权限

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
async def set_permission(ws, session_id, permission_name, setting="granted"):
    """
    设置浏览器权限
    - permission_name: 权限名称（geolocation, camera, microphone, notifications, etc.）
    - setting: granted, denied, prompt
    """
    return await cdp(ws, session_id, "Browser.setPermission", {
        "permission": {"name": permission_name},
        "setting": setting
    })


# 常用权限设置
async def grant_common_permissions(ws, session_id):
    """授予常用的浏览器权限"""
    permissions = [
        "geolocation",
        "camera",
        "microphone",
        "notifications",
        "midi",
        "midiSysex",
        "clipboardRead",
        "clipboardWrite",
    ]
    
    for perm in permissions:
        await set_permission(ws, session_id, perm, "granted")
    
    print(f"已授予 {len(permissions)} 个权限")


async def reset_permissions(ws, session_id):
    """重置所有权限为默认状态"""
    return await cdp(ws, session_id, "Browser.resetPermissions")


# 为特定域名设置权限
async def set_permission_for_origin(ws, session_id, origin, permission_name, setting="granted"):
    """为指定域名设置权限"""
    return await cdp(ws, session_id, "Browser.setPermission", {
        "origin": origin,
        "permission": {"name": permission_name},
        "setting": setting
    })

权限列表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
"""
CDP 支持的权限名称：
- geolocation        - 地理位置
- camera             - 摄像头
- microphone         - 麦克风
- notifications      - 通知
- midi               - MIDI
- midiSysex          - MIDI SysEx
- clipboardRead      - 剪贴板读取
- clipboardWrite     - 剪贴板写入
- backgroundSync     - 后台同步
- backgroundFetch    - 后台获取
- persistentStorage  - 持久存储
- push               - 推送通知
- vibrate            - 振动
- bluetooth          - 蓝牙
- usb                - USB
- windowPlacement    - 窗口放置
"""

实战：自动化安全测试

综合运用以上技术，构建自动化安全扫描器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def security_audit(ws, session_id, url):
    """
    对页面进行安全审计
    返回结构化安全报告
    """
    print(f"开始安全审计: {url}")
    report = {
        "url": url,
        "security_state": None,
        "mixed_content": [],
        "csp_violations": [],
        "certificate_issues": [],
        "permissions": {}
    }
    
    # 1. 启用所有监控
    await enable_security_monitoring(ws, session_id)
    await cdp(ws, session_id, "Network.enable")
    await cdp(ws, session_id, "Console.enable")
    
    # 2. 收集事件
    events = []
    
    async def collector():
        while True:
            try:
                msg = await asyncio.wait_for(ws.__anext__(), timeout=0.5)
                data = json.loads(msg)
                events.append(data)
            except asyncio.TimeoutError:
                break
            except Exception:
                break
    
    collector_task = asyncio.create_task(collector())
    
    # 3. 导航到页面
    await cdp(ws, session_id, "Page.navigate", {"url": url})
    await asyncio.sleep(5)
    
    collector_task.cancel()
    
    # 4. 分析结果
    for data in events:
        method = data.get("method", "")
        params = data.get("params", {})
        
        if method == "Security.securityStateChanged":
            report["security_state"] = params.get("securityState")
            
            for exp in params.get("explanations", []):
                if "mixed" in exp.get("summary", "").lower():
                    report["mixed_content"].append(exp["summary"])
        
        elif method == "Network.loadingFailed":
            if params.get("blockedReason") in ("csp", "mixed-content"):
                report["csp_violations"].append({
                    "url": params.get("url"),
                    "reason": params.get("blockedReason")
                })
        
        elif method == "Console.messageAdded":
            text = params.get("message", {}).get("text", "")
            if "CSP" in text:
                report["csp_violations"].append({"text": text[:100]})
    
    # 5. 检查页面安全状态
    print(f"\n=== 安全审计报告 ===")
    print(f"URL: {url}")
    print(f"安全状态: {report['security_state']}")
    print(f"混合内容: {len(report['mixed_content'])} 条")
    print(f"CSP 违规: {len(report['csp_violations'])} 条")
    
    return report

自动化证书过期检查

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
async def check_certificate_info(ws, session_id, url):
    """检查目标网站的证书信息"""
    # 导航到页面
    await cdp(ws, session_id, "Page.navigate", {"url": url})
    await asyncio.sleep(3)
    
    # 通过 Security 域获取安全信息
    await enable_security_monitoring(ws, session_id)
    
    # 监听安全状态
    cert_info = {}
    
    for _ in range(10):
        try:
            msg = await asyncio.wait_for(ws.__anext__(), timeout=1)
            data = json.loads(msg)
            
            if data.get("method") == "Security.securityStateChanged":
                p = data["params"]
                cert_info = {
                    "state": p.get("securityState"),
                    "explanations": [
                        e.get("summary") for e in p.get("explanations", [])
                    ]
                }
                break
        except asyncio.TimeoutError:
            continue
    
    return cert_info

常见踩坑与最佳实践

踩坑 1：setIgnoreCertificateErrors 必须在导航前设置

1
2
3
4
5
6
7
# ❌ 导航后才设置
await cdp(ws, session_id, "Page.navigate", {"url": "https://self-signed.bad"})
await ignore_certificate_errors(ws, session_id, True)  # 太晚了

# ✅ 先设置再导航
await ignore_certificate_errors(ws, session_id, True)
await cdp(ws, session_id, "Page.navigate", {"url": "https://self-signed.bad"})

踩坑 2：权限设置需要 HTTPS

大多数权限 API 要求在 HTTPS 页面下才能调用：

1
2
3
4
5
# ❌ HTTP 页面可能无法授予地理位置权限
await set_permission(ws, session_id, "geolocation", "granted")
# 但页面 navigator.permissions.query() 仍可能返回 denied

# ✅ 确保测试在 HTTPS 页面

踩坑 3：Security.enable 会触发 securityStateChanged

1
2
# 启用 Security 后，会立即收到当前页面的安全状态
# 对于新页面，会在导航完成后触发

踩坑 4：证书错误处理有两种方式

1
2
3
4
5
# 方式一：忽略所有证书错误（推荐用于测试）
await cdp(ws, session_id, "Security.setIgnoreCertificateErrors", {"ignore": True})

# 方式二：逐个处理证书错误事件
# 监听 Security.certificateError 并调用 Security.handleCertificateError

最佳实践清单

注意点 建议
证书忽略 导航前设置 setIgnoreCertificateErrors
权限授予 SSL 页面才能生效
CSP 监控 Console + Network 域同时监控
混合内容 Security.securityStateChanged 中包含
权限重置 测试结束后调用 resetPermissions
安全审计 组合 Security + Network + Console 域

完整参考：CDP 安全管理类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class CDPSecurityManager:
    def __init__(self, ws, session_id):
        self.ws = ws; self.session_id = session_id; self._cmd_id = 0
    
    async def _cmd(self, method, params=None):
        self._cmd_id += 1
        await self.ws.send(json.dumps({
            "sessionId": self.session_id, "id": self._cmd_id,
            "method": method, "params": params or {}
        }))
        async for resp in self.ws:
            data = json.loads(resp)
            if data.get("id") == self._cmd_id:
                return data.get("result", {})
    
    async def ignore_cert_errors(self, ignore=True):
        await self._cmd("Security.setIgnoreCertificateErrors", {"ignore": ignore})
    
    async def set_permission(self, name, setting="granted", origin=None):
        params = {"permission": {"name": name}, "setting": setting}
        if origin: params["origin"] = origin
        await self._cmd("Browser.setPermission", params)
    
    async def reset_permissions(self):
        await self._cmd("Browser.resetPermissions")
    
    async def audit(self, url):
        await self.ignore_cert_errors(True)
        await self._cmd("Security.enable")
        await self._cmd("Network.enable")
        
        await self._cmd("Page.navigate", {"url": url})
        await asyncio.sleep(3)
        
        issues = []
        try:
            while True:
                msg = await asyncio.wait_for(ws.__anext__(), timeout=2)
                data = json.loads(msg)
                m = data.get("method", "")
                if m in ("Security.securityStateChanged", "Network.loadingFailed",
                         "Console.messageAdded"):
                    issues.append(data)
        except (asyncio.TimeoutError, Exception):
            pass
        
        return issues

使用示例：

1
2
3
4
5
6
7
8
9
async with websockets.connect(CDP_URL) as ws:
    session_id = await connect_page(ws)
    sec = CDPSecurityManager(ws, session_id)
    
    await sec.set_permission("geolocation", "granted")
    await sec.ignore_cert_errors(True)
    
    issues = await sec.audit("https://example.com")
    print(f"发现 {len(issues)} 个安全问题")

总结：CDP 的 Security 和 Browser 域让你编程管理浏览器的安全策略——忽略证书错误、监控混合内容和 CSP 违规、授予和重置权限。这在自动化测试和安全审计中非常有用。

上一篇回顾：CDP Frame 管理指南：用 Python 处理 iframe 与跨域框架。

下一篇预告：CDP WebSocket 调试指南：用 Python 拦截与检查 WebSocket 帧。