爬虫实战项目分享

# DrissionPage 实现账号密码代理认证的完整指南

## 背景

DrissionPage 是一款基于 Python 的浏览器自动化工具，它封装了 Chromium 浏览器的操作。然而，DrissionPage 原生并不直接支持带账号密码认证的代理设置。当我们使用需要认证的代理时，浏览器会弹出认证对话框，导致自动化流程中断。

本文将介绍如何通过 Chromium DevTools Protocol (CDP) 的 Fetch API 来实现代理的账号密码自动认证。



## 核心思路

Chrome 浏览器提供了 CDP 接口，可以拦截网络请求并处理认证挑战。我们需要：

1. 启用 Fetch 域的事件监听
2. 注册 `Fetch.authRequired` 回调来处理认证请求
3. 在回调中提供代理的账号密码

## 实现步骤

### 1. 解析代理 URL

首先，我们需要从代理 URL 中提取出协议、主机、端口、用户名和密码等信息。

```python
from urllib.parse import urlparse
from dataclasses import dataclass
from typing import Optional


@dataclass
class Proxy:
    scheme: str
    host: str
    port: Optional[int] = None
    username: Optional[str] = None
    password: Optional[str] = None

    @classmethod
    def from_url(cls, proxy_url: str):
        """从代理 URL 解析出各个组件"""
        parsed = urlparse(proxy_url)

        return cls(
            scheme=parsed.scheme,
            host=parsed.hostname or "",
            port=parsed.port,
            username=parsed.username,
            password=parsed.password,
        )

    @property
    def url(self) -> str:
        """生成不带认证的代理 URL，用于浏览器配置"""
        host = self.host
        if self.port:
            host += f":{self.port}"
        return f"{self.scheme}://{host}"
```

使用示例：

```python
proxy = Proxy.from_url("http://user:pass@proxy.example.com:8080")

print(proxy.scheme)    # http
print(proxy.host)      # proxy.example.com
print(proxy.port)      # 8080
print(proxy.username)  # user
print(proxy.password)  # pass
print(proxy.url)       # http://proxy.example.com:8080
```

### 2. 注册认证回调

这是实现代理认证的核心部分。我们需要在浏览器启动后注册 `Fetch.authRequired` 事件的回调函数。

```python
from DrissionPage import ChromiumPage, ChromiumOptions
from typing import Optional


class BrowserClient:
    def __init__(self, proxy_url: str = None):
        self.driver: Optional[ChromiumPage] = None
        self.proxy_url = proxy_url
        self._proxy = None

    def _on_auth_required(self, **params):
        """处理代理认证请求"""
        try:
            request_id = params.get('requestId')
            auth_challenge = params.get('authChallenge', {})

            if auth_challenge.get('source') == 'Proxy' and self._proxy:
                # 提供代理认证信息
                self.driver.run_cdp(
                    'Fetch.continueWithAuth',
                    requestId=request_id,
                    authChallengeResponse={
                        'response': 'ProvideCredentials',
                        'username': self._proxy.username,
                        'password': self._proxy.password
                    }
                )
            else:
                # 使用默认认证响应
                self.driver.run_cdp(
                    'Fetch.continueWithAuth',
                    requestId=request_id,
                    authChallengeResponse={'response': 'Default'}
                )
        except Exception:
            pass

    def _on_request_paused(self, **params):
        """处理请求暂停事件，必须继续请求"""
        try:
            request_id = params.get('requestId')
            self.driver.run_cdp(
                'Fetch.continueRequest',
                requestId=request_id,
            )
        except Exception:
            pass
```

### 3. 启动浏览器并启用拦截

在浏览器启动后、访问任何页面之前，我们需要启用 Fetch API 并注册回调。

```python
    def launch_browser(self):
        co = ChromiumOptions()

        # 配置代理（不带认证信息的 URL）
        if self.proxy_url:
            self._proxy = Proxy.from_url(self.proxy_url)
            co.set_proxy(self._proxy.url)

        # 启动浏览器
        self.driver = ChromiumPage(co)

        # 先打开一个空白页面
        self.driver.get("about:blank")

        # 注册回调
        self.driver.driver.set_callback('Fetch.authRequired', self._on_auth_required)
        self.driver.driver.set_callback('Fetch.requestPaused', self._on_request_paused)

        # 启用 Fetch 域，handleAuthRequests=True 是关键
        self.driver.run_cdp('Fetch.enable', handleAuthRequests=True)
```

### 4. 完整的封装类

将以上代码组合起来，加上浏览器管理方法：

```python
from DrissionPage import ChromiumPage, ChromiumOptions
from urllib.parse import urlparse
from dataclasses import dataclass
from typing import Optional


@dataclass
class Proxy:
    scheme: str
    host: str
    port: Optional[int] = None
    username: Optional[str] = None
    password: Optional[str] = None

    @classmethod
    def from_url(cls, proxy_url: str):
        parsed = urlparse(proxy_url)
        return cls(
            scheme=parsed.scheme,
            host=parsed.hostname or "",
            port=parsed.port,
            username=parsed.username,
            password=parsed.password,
        )

    @property
    def url(self) -> str:
        host = self.host
        if self.port:
            host += f":{self.port}"
        return f"{self.scheme}://{host}"


class BrowserClient:
    def __init__(self, proxy_url: str = None):
        self.driver: Optional[ChromiumPage] = None
        self.proxy_url = proxy_url
        self._proxy = None

    def _on_auth_required(self, **params):
        try:
            request_id = params.get('requestId')
            auth_challenge = params.get('authChallenge', {})

            if auth_challenge.get('source') == 'Proxy' and self._proxy:
                self.driver.run_cdp(
                    'Fetch.continueWithAuth',
                    requestId=request_id,
                    authChallengeResponse={
                        'response': 'ProvideCredentials',
                        'username': self._proxy.username,
                        'password': self._proxy.password
                    }
                )
            else:
                self.driver.run_cdp(
                    'Fetch.continueWithAuth',
                    requestId=request_id,
                    authChallengeResponse={'response': 'Default'}
                )
        except Exception:
            pass

    def _on_request_paused(self, **params):
        try:
            request_id = params.get('requestId')
            self.driver.run_cdp('Fetch.continueRequest', requestId=request_id)
        except Exception:
            pass

    def launch_browser(self):
        co = ChromiumOptions()

        if self.proxy_url:
            self._proxy = Proxy.from_url(self.proxy_url)
            co.set_proxy(self._proxy.url)

        self.driver = ChromiumPage(co)
        self.driver.get("about:blank")

        self.driver.driver.set_callback('Fetch.authRequired', self._on_auth_required)
        self.driver.driver.set_callback('Fetch.requestPaused', self._on_request_paused)
        self.driver.run_cdp('Fetch.enable', handleAuthRequests=True)

    def get_tab(self) -> ChromiumPage:
        if not self.driver:
            self.launch_browser()
        return self.driver

    def close_browser(self):
        if self.driver:
            try:
                self.driver.quit()
            except Exception:
                pass
        self.driver = None
```

## 使用示例

### 基本用法

```python
# 创建客户端
client = BrowserClient(proxy_url="http://user:password@proxy.example.com:8080")

# 启动浏览器（自动处理认证）
tab = client.get_tab()

# 访问网站
tab.get("https://httpbin.org/ip")
print(tab.eles('tag:pre')[0].text)

# 关闭浏览器
client.close_browser()
```

### 支持多种代理格式

```python
# HTTP 代理
client = BrowserClient(proxy_url="http://user:pass@proxy.com:8080")

# SOCKS5 代理
client = BrowserClient(proxy_url="socks5://user:pass@proxy.com:1080")

# 无认证代理
client = BrowserClient(proxy_url="http://proxy.com:8080")

# 无代理
client = BrowserClient(proxy_url=None)
```

### 在爬虫中使用

```python
def crawl_with_proxy(url, proxy_url):
    client = BrowserClient(proxy_url=proxy_url)
    try:
        tab = client.get_tab()
        tab.get(url)
        # 执行你的爬虫逻辑
        title = tab.title
        print(f"页面标题: {title}")
        return tab.html
    finally:
        client.close_browser()


# 使用示例
html = crawl_with_proxy(
    url="https://httpbin.org/ip",
    proxy_url="http://user:password@proxy.example.com:8080"
)
```

## 关键要点说明

### 1. 为什么需要 `Fetch.enable`？

Chrome 的 Fetch 域允许我们拦截网络请求。设置 `handleAuthRequests=True` 后，当浏览器遇到需要认证的请求时，会触发 `Fetch.authRequired` 事件，给我们提供处理认证的机会。

### 2. 为什么需要 `Fetch.requestPaused` 回调？

启用 Fetch 拦截后，所有请求都会被暂停。如果我们不处理这些暂停的请求，页面将无法加载任何资源。因此需要注册一个回调函数来继续这些请求。

### 3. 为什么先访问 `about:blank`？

我们需要在浏览器启动后、访问目标网站之前注册回调函数。访问 `about:blank` 可以确保浏览器已经初始化完成，同时不会触发任何外部请求。

### 4. `authChallenge.source` 的判断

认证挑战可能来自多个来源（Proxy、Server、Proxy 等）。我们需要检查 `source == 'Proxy'` 来确保我们只在处理代理认证时提供账号密码。

## 常见问题

**Q: 认证仍然失败怎么办？**

检查代理 URL 格式是否正确，确保用户名和密码没有特殊字符编码问题。

**Q: 页面加载很慢？**

这通常是代理服务器本身的问题，与认证实现无关。可以尝试其他代理测试。

**Q: 可以动态切换代理吗？**

不建议在同一个浏览器会话中切换代理。更好的做法是关闭当前浏览器，创建新的客户端实例。

## 总结

通过 CDP 的 Fetch API，我们可以在 DrissionPage 中优雅地处理需要账号密码认证的代理。核心步骤是：

1. 解析代理 URL 获取认证信息
2. 使用不带认证的代理 URL 配置浏览器
3. 启用 Fetch 域并注册认证回调
4. 在回调中提供账号密码完成认证

这种方法不仅适用于 DrissionPage，也可以应用于其他使用 CDP 的浏览器自动化工具。

DrissonPage账密认证代理

某皮 web端设备指纹算法分析

某皮ios和安卓sap算法分析

# 固定时间戳
currentTimeMillis
```java
protected long currentTimeMillis() {
    //return System.currentTimeMillis();
    System.out.println("固定时间戳:" + 1744706898936L);
    return 1744706898936L;
}
```



clock_gettime

```java
private static final int CLOCK_REALTIME = 0;
    private static final int CLOCK_MONOTONIC = 1;
    private static final int CLOCK_THREAD_CPUTIME_ID = 3;
    private static final int CLOCK_MONOTONIC_RAW = 4;
    private static final int CLOCK_MONOTONIC_COARSE = 6;
    private static final int CLOCK_BOOTTIME = 7;

    private int CLOCK_TIMES_0 = 0;
    private int CLOCK_TIMES_1 = 0;

    private final long nanoTime = System.nanoTime();

    List<Long> list0 = new ArrayList<Long>() {{
        add(1765768512725878000L);
        add(1765768512823353000L);
        add(1765768512916020000L);
        add(1765768513012195000L);
        add(1765768513108126000L);
    }};
    List<Long> list1 = new ArrayList<Long>() {{
        add(18136934846099106L);
        add(18136934848744287L);
        add(18136934849180744L);
        add(18136934850204595L);
        add(18136934850400232L);
        add(18136934850362437L);
        add(18136934850687493L);
    }};

    protected int clock_gettime(Emulator<?> emulator) {
        RegisterContext context = emulator.getContext();
        int clk_id = context.getIntArg(0) & 0x7;
        Pointer tp = context.getPointerArg(1);
        long offset0 = clk_id == CLOCK_REALTIME ? currentTimeMillis() * 1000000L : System.nanoTime() - nanoTime;
        if (clk_id == 0) {
            CLOCK_TIMES_0 += 1;
        } else if (clk_id == 1) {
            CLOCK_TIMES_1 += 1;
        }
        long offset;
        if (CLOCK_TIMES_0 <= 5 && clk_id == 0) {
            offset = list0.get(CLOCK_TIMES_0 - 1);
        } else if (CLOCK_TIMES_1 <= 10 && clk_id == 1) {
            offset = list1.get(CLOCK_TIMES_1 - 1);
        } else {
            offset = offset0;
        }

        System.out.println("\n==========> offset0= " + offset0 + "==============\n");

        long tv_sec = offset / 1000000000L;
        long tv_nsec = offset % 1000000000L;
        if (log.isDebugEnabled()) {
            log.debug("clock_gettime clk_id={}, tp={}, offset={}, tv_sec={}, tv_nsec={}", clk_id, tp, offset, tv_sec, tv_nsec);
        }
        switch (clk_id) {
            case CLOCK_REALTIME:
            case CLOCK_MONOTONIC:
            case CLOCK_THREAD_CPUTIME_ID:
            case CLOCK_MONOTONIC_RAW:
            case CLOCK_MONOTONIC_COARSE:
            case CLOCK_BOOTTIME:
                tp.setLong(0, tv_sec);
                tp.setLong(8, tv_nsec);
                return 0;
        }
        if (log.isDebugEnabled()) {
            emulator.attach().debug();
        }
        throw new UnsupportedOperationException("clk_id=" + clk_id);
    }
```

# 固定randBytes(/dev/urandom)

```java
package com.github.unidbg.linux.file;

import com.github.unidbg.Emulator;
import com.github.unidbg.arm.backend.Backend;
import com.github.unidbg.file.linux.IOConstants;
import com.sun.jna.Pointer;

import java.util.Arrays;
import java.util.concurrent.ThreadLocalRandom;

public class RandomFileIO extends DriverFileIO {

    public RandomFileIO(Emulator<?> emulator, String path) {
        super(emulator, IOConstants.O_RDONLY, path);
    }

    public static String toHex(byte[] bytes) {
        StringBuilder hexString = new StringBuilder();
        for (byte b : bytes) {
            hexString.append(String.format("%02x ", b));
        }
        return hexString.toString().trim();

    }

    @Override
    public int read(Backend backend, Pointer buffer, int count) {
        int total = 0;
        byte[] buf = new byte[Math.min(0x1000, count)];
        randBytes(buf);
        System.out.println("固定随机数:" + toHex(buf));
        Pointer pointer = buffer;
        while (total < count) {
            int read = Math.min(buf.length, count - total);
            pointer.write(0, buf, 0, read);
            total += read;
            pointer = pointer.share(read);
        }
        return total;

    }

    private static final byte[] FIXED_RANDOM_BYTES = {
            (byte) 0x12, (byte) 0x34, (byte) 0x56, (byte) 0x78,
            (byte) 0x9A, (byte) 0xBC, (byte) 0xDE, (byte) 0xF0
    };

    protected void randBytes(byte[] bytes) {
        if (bytes == null || bytes.length == 0) {
            return;
        }

        for (int i = 0; i < bytes.length; i++) {
            bytes[i] = FIXED_RANDOM_BYTES[i % FIXED_RANDOM_BYTES.length];
        }
    }

}
```

爬虫项目实战

Star History

DrissionPage 实现账号密码代理认证的完整指南

背景

某皮网页端设备指纹 LongToken 技术深度分析

1. 概述

某皮移动端反爬虫算法深度解析：Android 与 iOS 平台对比分析

一、概述

固定时间戳