全球动态代理 · 静态住宅代理 · 极高质量 · 包月不限量 · 学员低价 · 联系如意进行免费测试 · 欢迎公司及学员采购 · 另有任何海外数据采集需求，随时联系 · 商务合作微信：Charleval
Automation / 认证
RuyiPage Docs

RuyiPage 认证

当服务器返回 HTTP 401 认证质询时，浏览器触发 authRequired 事件。通过拦截该事件并调用 continue_with_auth 提供凭证，无需弹窗交互即可完成认证。

认证流程

  • 浏览器访问需要 HTTP Basic/Digest 认证的资源。
  • 服务器返回 401，浏览器触发 authRequired 事件。
  • RuyiPage 拦截该事件，自动提供用户名和密码。
  • 服务器验证通过，返回正常响应。

示例：自动处理 HTTP 认证

python 拦截 authRequired 并提供凭证 
from ruyipage import FirefoxPage

page = FirefoxPage()

# 订阅 authRequired 事件（用于确认认证触发）
page.events.start(["network.authRequired"], contexts=[page.tab_id])

# 拦截 authRequired 阶段
page.intercept.start(
    url_patterns=[{"type": "string", "pattern": "https://example.com/api/auth"}],
    phases=["authRequired"]
)

# 访问需要认证的页面
try:
    page.get("https://example.com/api/auth")
except Exception:
    pass

# 等待认证请求
auth_req = page.intercept.wait(timeout=5)
auth_event = page.events.wait("network.authRequired", timeout=5)

if auth_req and auth_event:
    # 提供用户名和密码
    auth_req.continue_with_auth(
        action="provideCredentials",
        username="admin",
        password="secret123"
    )
    print("认证凭证已提供")

    # 等待认证后的响应
    done = page.events.wait("network.responseCompleted", timeout=5)
    if done:
        status = (done.response or {}).get("status", "")
        print(f"认证结果: status={status}")

page.intercept.stop()
page.events.stop()
page.quit()

continue_with_auth 的三种 action

action含义场景
provideCredentials提供用户名和密码最常用，自动登录
cancel取消认证放弃访问该资源
default使用浏览器默认行为弹出浏览器原生认证对话框

重点 API 参考

req.continue_with_auth

返回: None
req.continue_with_auth(action='provideCredentials', username=None, password=None)

处理 HTTP 认证质询，提供凭证或执行其他认证动作。

参数说明
参数说明
action认证动作：provideCredentials / cancel / default。
username用户名（action=provideCredentials 时必填）。
password密码（action=provideCredentials 时必填）。
常见坑
  • 仅在 authRequired 阶段的拦截请求上可调用。
  • 如果凭证错误，浏览器会再次触发 authRequired。
相关示例
网络事件：请求开始、响应开始、响应完成、错误与认证剩余命令覆盖：截图、历史、继续响应、mock、fail、认证、动作释放

相关示例源码

2 个示例 ▶ 点击展开

网络事件：请求开始、响应开始、响应完成、错误与认证

examples/31_network_events.py

是 network-events 页最直接的主示例。

GitHub 仓库 查看仓库源码 page.events.startpage.events.waitpage.intercept.startpage.intercept.waitcontinue_with_auth
  • 观察 beforeRequestSent、responseStarted、responseCompleted
  • 监听 fetchError
  • 处理 authRequired 并补充用户名密码

源码文件：app/content/ruyipage_examples/31_network_events.py · 仓库：31_network_events.py

python 注释版 
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 网络事件：请求开始、响应开始、响应完成、错误与认证
# 说明: 是 network-events 页最直接的主示例。
# 你会学到:
#   1. 观察 beforeRequestSent、responseStarted、responseCompleted
#   2. 监听 fetchError
#   3. 处理 authRequired 并补充用户名密码
# 相关 API: page.events.start, page.events.wait, page.intercept.start, page.intercept.wait, continue_with_auth
# =============================

# -*- coding: utf-8 -*-
"""示例31: Network Events（严格结果版）"""

import io
import sys
from typing import Dict, List

if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")

from ruyipage import FirefoxPage, InterceptedRequest
from ruyipage._functions.tools import find_free_port

from test_server import TestServer


def add_result(
    results: List[Dict[str, str]], item: str, status: str, note: str
) -> None:
    results.append({"item": item, "status": status, "note": note})


def print_results(results: List[Dict[str, str]]) -> None:
    print("\n| 项目 | 状态 | 说明 |")
    print("| --- | --- | --- |")
    for row in results:
        print(f"| {row['item']} | {row['status']} | {row['note']} |")


def main() -> None:
    print("=" * 70)
    print("测试 31: Network Events")
    print("=" * 70)

    page = FirefoxPage()
    server = None
    results: List[Dict[str, str]] = []

    try:
        server = TestServer(port=find_free_port(9330, 9430)).start()
        base_url = server.get_url("")[:-1]

        page.get("about:blank")
        add_result(
            results, "页面加载", "成功", f"about:blank 已加载，服务地址: {base_url}"
        )

        page.events.start(
            [
                "network.beforeRequestSent",
                "network.responseStarted",
                "network.responseCompleted",
                "network.fetchError",
                "network.authRequired",
            ],
            contexts=[page.tab_id],
        )

        # 1. 普通请求事件链
        page.events.clear()
        page.run_js(
            f"fetch('{base_url}/api/data').catch(()=>null); return true;",
            as_expr=False,
        )

        before_event = page.events.wait("network.beforeRequestSent", timeout=5)
        started_event = page.events.wait("network.responseStarted", timeout=5)
        completed_event = page.events.wait("network.responseCompleted", timeout=5)

        if before_event and str((before_event.request or {}).get("url", "")).endswith(
            "/api/data"
        ):
            add_result(
                results,
                "network.beforeRequestSent",
                "成功",
                (before_event.request or {}).get("url", ""),
            )
        else:
            add_result(
                results,
                "network.beforeRequestSent",
                "失败",
                "未观察到 /api/data 请求开始事件",
            )

        if (
            started_event
            and int((started_event.response or {}).get("status", 0)) == 200
        ):
            add_result(
                results,
                "network.responseStarted",
                "成功",
                f"status={(started_event.response or {}).get('status')}",
            )
        else:
            add_result(
                results, "network.responseStarted", "失败", "未观察到 200 响应开始事件"
            )

        if (
            completed_event
            and int((completed_event.response or {}).get("status", 0)) == 200
        ):
            add_result(
                results,
                "network.responseCompleted",
                "成功",
                f"status={(completed_event.response or {}).get('status')}",
            )
        else:
            add_result(
                results,
                "network.responseCompleted",
                "失败",
                "未观察到 200 响应完成事件",
            )

        # 2. fetchError
        page.events.clear()
        page.run_js(
            "fetch('http://127.0.0.1:9/').catch(()=>null); return true;",
            as_expr=False,
        )
        error_event = page.events.wait("network.fetchError", timeout=5)
        if error_event:
            add_result(
                results, "network.fetchError", "成功", error_event.error_text or ""
            )
        else:
            add_result(
                results, "network.fetchError", "跳过", "当前环境未稳定观察到 fetchError"
            )

        # 3. authRequired
        page.events.clear()
        page.intercept.start(
            url_patterns=[{"type": "string", "pattern": base_url + "/api/auth"}],
            phases=["authRequired"],
        )
        try:
            page.get(base_url + "/api/auth")
        except Exception:
            pass

        auth_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
        auth_event = page.events.wait("network.authRequired", timeout=5)

        if auth_event and auth_req:
            add_result(results, "network.authRequired", "成功", auth_req.url)
            auth_req.continue_with_auth(
                action="provideCredentials",
                username="user",
                password="pass",
            )

            auth_done = page.events.wait("network.responseCompleted", timeout=5)
            if auth_done and int((auth_done.response or {}).get("status", 0)) == 200:
                add_result(
                    results,
                    "network.authRequired credentials",
                    "成功",
                    "提供凭证后认证通过",
                )
            else:
                add_result(
                    results,
                    "network.authRequired credentials",
                    "失败",
                    "提供凭证后未观察到 200 完成事件",
                )
        else:
            add_result(
                results,
                "network.authRequired",
                "跳过",
                "当前环境未稳定观察到 authRequired",
            )
            add_result(
                results,
                "network.authRequired credentials",
                "跳过",
                "未观察到 authRequired，跳过凭证验证",
            )

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.intercept.stop()
        except Exception:
            pass
        try:
            page.events.stop()
        except Exception:
            pass
        try:
            if server is not None:
                server.stop()
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

剩余命令覆盖：截图、历史、继续响应、mock、fail、认证、动作释放

examples/34_remaining_commands.py

把一些散落但重要的能力集中演示，适合做协议能力查漏补缺。

GitHub 仓库 查看仓库源码 page.screenshotpage.backpage.forwardpage.intercept.start_responsesreq.continue_responsereq.mockreq.failreq.continue_with_auth
  • 截图与保存文件
  • 验证历史遍历
  • 继续响应和 mock 响应
  • 请求失败与认证补充
  • 释放动作链状态

源码文件：app/content/ruyipage_examples/34_remaining_commands.py · 仓库：34_remaining_commands.py

python 注释版 
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 剩余命令覆盖：截图、历史、继续响应、mock、fail、认证、动作释放
# 说明: 把一些散落但重要的能力集中演示，适合做协议能力查漏补缺。
# 你会学到:
#   1. 截图与保存文件
#   2. 验证历史遍历
#   3. 继续响应和 mock 响应
#   4. 请求失败与认证补充
#   5. 释放动作链状态
# 相关 API: page.screenshot, page.back, page.forward, page.intercept.start_responses, req.continue_response, req.mock, req.fail, req.continue_with_auth
# =============================

# -*- coding: utf-8 -*-
"""示例34: Remaining Commands（严格结果版）"""

import io
import os
import sys
from typing import Dict, List

if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")

from ruyipage import FirefoxPage, InterceptedRequest
from ruyipage._functions.tools import find_free_port

from test_server import TestServer


def add_result(
    results: List[Dict[str, str]], item: str, status: str, note: str
) -> None:
    results.append({"item": item, "status": status, "note": note})


def print_results(results: List[Dict[str, str]]) -> None:
    print("\n| 项目 | 状态 | 说明 |")
    print("| --- | --- | --- |")
    for row in results:
        print(f"| {row['item']} | {row['status']} | {row['note']} |")


def main() -> None:
    print("=" * 70)
    print("测试 34: Remaining Commands")
    print("=" * 70)

    page = FirefoxPage()
    server = None
    results: List[Dict[str, str]] = []
    screenshot_path = os.path.join(os.path.dirname(__file__), "test_screenshot.png")

    try:
        server = TestServer(port=find_free_port(9430, 9530)).start()
        base_url = server.get_url("")[:-1]

        # 1. captureScreenshot
        page.get(base_url + "/nav/basic")
        add_result(results, "页面加载", "成功", f"服务地址: {base_url}")

        data_bytes = page.screenshot(as_bytes=True)
        if data_bytes and len(data_bytes) > 100:
            add_result(
                results,
                "browsingContext.captureScreenshot viewport",
                "成功",
                f"截图字节数: {len(data_bytes)}",
            )
        else:
            add_result(
                results,
                "browsingContext.captureScreenshot viewport",
                "失败",
                "截图数据过短",
            )

        saved_path = page.screenshot(path=screenshot_path)
        if saved_path and os.path.exists(saved_path):
            add_result(results, "screenshot save file", "成功", saved_path)
        else:
            add_result(results, "screenshot save file", "失败", "未保存截图文件")

        # 2. traverseHistory
        page.get(base_url + "/nav/basic?a=1")
        page.get(base_url + "/nav/basic?a=2")
        page.get(base_url + "/nav/basic?a=3")
        page.back()
        back_url = page.url
        page.back()
        back2_url = page.url
        page.forward()
        forward_url = page.url

        if (
            back_url.endswith("?a=2")
            and back2_url.endswith("?a=1")
            and forward_url.endswith("?a=2")
        ):
            add_result(
                results,
                "browsingContext.traverseHistory",
                "成功",
                f"back={back_url}, back2={back2_url}, forward={forward_url}",
            )
        else:
            add_result(
                results,
                "browsingContext.traverseHistory",
                "失败",
                f"back={back_url}, back2={back2_url}, forward={forward_url}",
            )

        # 3. network.continueResponse
        page.intercept.start_responses(
            url_patterns=[{"type": "string", "pattern": base_url + "/api/data"}]
        )
        page.run_js(
            f"fetch('{base_url}/api/data').catch(()=>null); return true;", as_expr=False
        )
        resp_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
        if resp_req:
            resp_req.continue_response()
            add_result(results, "network.continueResponse", "成功", resp_req.url)
        else:
            add_result(
                results,
                "network.continueResponse",
                "跳过",
                "未稳定捕获到 responseStarted 拦截",
            )
        page.intercept.stop()

        # 4. network.provideResponse
        page.intercept.start_requests(
            url_patterns=[{"type": "string", "pattern": base_url + "/api/mock-source"}]
        )
        page.run_js(
            f"fetch('{base_url}/api/mock-source').catch(()=>null); return true;",
            as_expr=False,
        )
        mock_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
        if mock_req:
            mock_req.mock(body='{"status":"mocked"}', status_code=200)
            add_result(results, "network.provideResponse", "成功", mock_req.url)
        else:
            add_result(
                results,
                "network.provideResponse",
                "跳过",
                "未稳定捕获到 beforeRequestSent 拦截",
            )
        page.intercept.stop()

        # 5. network.failRequest
        page.intercept.start_requests(
            url_patterns=[{"type": "string", "pattern": base_url + "/api/slow"}]
        )
        page.run_js(
            f"fetch('{base_url}/api/slow').catch(()=>null); return true;", as_expr=False
        )
        fail_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
        if fail_req:
            fail_req.fail()
            add_result(results, "network.failRequest", "成功", fail_req.url)
        else:
            add_result(
                results,
                "network.failRequest",
                "跳过",
                "未稳定捕获到 beforeRequestSent 拦截",
            )
        page.intercept.stop()

        # 6. network.continueWithAuth
        page.events.start(["network.authRequired"], contexts=[page.tab_id])
        page.intercept.start(
            url_patterns=[{"type": "string", "pattern": base_url + "/api/auth"}],
            phases=["authRequired"],
        )
        try:
            page.get(base_url + "/api/auth")
        except Exception:
            pass
        auth_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
        auth_event = page.events.wait("network.authRequired", timeout=5)
        if auth_req and auth_event:
            auth_req.continue_with_auth(
                action="provideCredentials",
                username="user",
                password="pass",
            )
            add_result(results, "network.continueWithAuth", "成功", auth_req.url)
        else:
            add_result(
                results,
                "network.continueWithAuth",
                "跳过",
                "当前环境未稳定观察到 authRequired",
            )
        page.intercept.stop()
        page.events.stop()

        # 7. input.performActions / releaseActions
        try:
            page.quit()
        except Exception:
            pass
        page = FirefoxPage()

        action_page = """
        <!DOCTYPE html>
        <html>
        <head><meta charset='utf-8'><title>Actions Test</title></head>
        <body>
            <input id='input-box' autofocus />
            <button id='btn' onclick="window.btnClicked = true">Click Me</button>
            <script>
                window.btnClicked = false;
                window.lastKey = '';
                document.getElementById('input-box').addEventListener('keydown', function(e) {
                    window.lastKey = e.key;
                });
            </script>
        </body>
        </html>
        """
        page.get("data:text/html;charset=utf-8," + action_page)
        page.actions.move_to(page.ele("#input-box")).click().press("a").perform()
        last_key = page.run_js("return window.lastKey")
        page.actions.move_to(page.ele("#btn")).click().perform()
        btn_clicked = page.run_js("return window.btnClicked")
        page.actions.release()

        if last_key == "a":
            add_result(
                results, "input.performActions keyboard", "成功", f"lastKey={last_key}"
            )
        else:
            add_result(
                results, "input.performActions keyboard", "失败", f"lastKey={last_key}"
            )

        if btn_clicked is True:
            add_result(
                results, "input.performActions pointer", "成功", "按钮点击事件已触发"
            )
        else:
            add_result(
                results, "input.performActions pointer", "失败", "按钮点击事件未触发"
            )

        add_result(results, "input.releaseActions", "成功", "动作状态已释放")

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.intercept.stop()
        except Exception:
            pass
        try:
            page.events.stop()
        except Exception:
            pass
        try:
            if os.path.exists(screenshot_path):
                os.remove(screenshot_path)
        except Exception:
            pass
        try:
            if server is not None:
                server.stop()
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()