回锅炒辣椒的blog
DrissonPage账密认证代理
2026-04-10
自动化
00

目录

DrissionPage 实现账号密码代理认证的完整指南
背景
核心思路
实现步骤
1. 解析代理 URL
2. 注册认证回调
3. 启动浏览器并启用拦截
4. 完整的封装类
使用示例
基本用法
支持多种代理格式
在爬虫中使用
关键要点说明
1. 为什么需要 Fetch.enable？
2. 为什么需要 Fetch.requestPaused 回调？
3. 为什么先访问 about:blank？
4. authChallenge.source 的判断
常见问题
总结

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 的浏览器自动化工具。

如果对你有用的话，可以打赏哦
打赏
ali pay
wechat pay

本文作者:回锅炒辣椒

本文链接:

版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！