---

name: performing-paste-site-monitoring-for-credentials description: 使用自动化抓取和关键词匹配技术，监控 Pastebin 和 GitHub Gists 等粘贴站点上的泄露凭证、API 密钥和敏感数据转储，实现早期泄露检测 domain: cybersecurity subdomain: threat-intelligence tags: [paste-monitoring, credential-leak, pastebin, data-breach, threat-intelligence, osint, early-warning] version: "1.0" author: mahipal license: Apache-2.0

执行粘贴站点凭证监控

概述

粘贴站点（Pastebin、GitHub Gists、Ghostbin、Dpaste、Hastebin）经常被用作泄露凭证（Credential）、数据库转储（Database Dump）、API 密钥和敏感数据在暗网论坛及 Telegram 频道广泛传播前的暂存区域。监控这些站点可实现早期泄露检测（Breach Detection），使组织能够在被盗数据被武器化之前及时响应。本技能涵盖使用 Pastebin Scraping API 构建自动化粘贴站点监控器、基于关键词的告警、凭证模式匹配，以及与事件响应（Incident Response）工作流的集成。

前置条件

• Python 3.9+，安装 requests、beautifulsoup4、regex、pymisp 库
• 具有 Scraping API 访问权限的 Pastebin PRO 账户（每月 $49.95，用于程序化访问）
• 用于 Gist 监控的 GitHub API 令牌
• 针对您组织的特定关键词列表（域名、项目名称、内部术语）
• 用于粘贴存储和搜索的 Elasticsearch 或数据库

核心概念

粘贴站点威胁态势

每年有超过 300,000 个用户凭证发布在 Pastebin 上，平均每次泄露包含 1,000 个用户名/密码对。粘贴站点服务于三个主要的威胁情报目的：早期泄露检测（凭证出现在粘贴站点上早于暗网）、威胁行为者画像（攻击者使用粘贴站点进行 C2 配置、数据暂存、工具共享）和恶意软件发现（编码的有效载荷、配置文件、C2 地址）。

监控方法

主动监控（Active Monitoring）定期查询粘贴站点 API 或抓取端点。Pastebin Scraping API 提供对新公开粘贴的实时访问。对于 GitHub，搜索 API 允许监控 Gists 和代码库提交中的暴露密钥。被动监控（Passive Monitoring）使用 IntelX、Dehashed 或 Have I Been Pwned 等聚合粘贴站点数据的服务。

凭证模式检测

有效的监控使用正则表达式（Regex）模式检测：电子邮件:密码组合、API 密钥（AWS、Azure、GCP、Stripe、Twilio）、数据库连接字符串（Connection String）、私钥（SSH、PGP）、JWT 令牌，以及内部主机名/URL。组织特定关键词（域名、产品名称、员工姓名）可降低误报率。

实操步骤

步骤 1：Pastebin Scraping API 监控器

import requests
import re
import json
import time
from datetime import datetime

class PastebinMonitor:
    SCRAPING_URL = "https://scrape.pastebin.com/api_scraping.php"
    RAW_URL = "https://scrape.pastebin.com/api_scrape_item.php"

    def __init__(self, keywords, output_dir="paste_alerts"):
        self.keywords = [k.lower() for k in keywords]
        self.output_dir = output_dir
        self.seen_keys = set()
        self.credential_patterns = {
            "email_password": re.compile(
                r'[\w.+-]+@[\w-]+\.[\w.]+[\s:;|,]+[\S]{6,}', re.IGNORECASE),
            "aws_key": re.compile(
                r'AKIA[0-9A-Z]{16}'),
            "aws_secret": re.compile(
                r'[0-9a-zA-Z/+=]{40}'),
            "github_token": re.compile(
                r'ghp_[0-9a-zA-Z]{36}'),
            "slack_token": re.compile(
                r'xox[baprs]-[0-9a-zA-Z-]+'),
            "private_key": re.compile(
                r'-----BEGIN (?:RSA |EC |DSA )?PRIVATE KEY-----'),
            "jwt_token": re.compile(
                r'eyJ[A-Za-z0-9-_]+\.eyJ[A-Za-z0-9-_]+\.[A-Za-z0-9-_]+'),
            "connection_string": re.compile(
                r'(?:mongodb|postgres|mysql|redis)://[^\s]+'),
            "api_key_generic": re.compile(
                r'(?:api[_-]?key|apikey|access[_-]?token)[\s]*[=:]\s*["\']?[\w-]{20,}',
                re.IGNORECASE),
        }

    def fetch_recent_pastes(self, limit=100):
        """从 Pastebin Scraping API 获取最近的公开粘贴。"""
        params = {"limit": limit}
        try:
            resp = requests.get(self.SCRAPING_URL, params=params, timeout=30)
            if resp.status_code == 200:
                pastes = resp.json()
                print(f"[+] 已获取 {len(pastes)} 条最新粘贴")
                return pastes
            else:
                print(f"[-] API 错误: {resp.status_code}")
                return []
        except Exception as e:
            print(f"[-] 获取失败: {e}")
            return []

    def get_paste_content(self, paste_key):
        """获取粘贴的原始内容。"""
        params = {"i": paste_key}
        try:
            resp = requests.get(self.RAW_URL, params=params, timeout=15)
            if resp.status_code == 200:
                return resp.text
            return ""
        except Exception:
            return ""

    def analyze_paste(self, content, paste_metadata):
        """分析粘贴内容中的凭证和关键词。"""
        findings = {
            "keyword_matches": [],
            "credential_matches": {},
            "severity": "low",
        }

        content_lower = content.lower()

        # 检查关键词
        for keyword in self.keywords:
            if keyword in content_lower:
                count = content_lower.count(keyword)
                findings["keyword_matches"].append({
                    "keyword": keyword,
                    "count": count,
                })

        # 检查凭证模式
        for pattern_name, pattern in self.credential_patterns.items():
            matches = pattern.findall(content)
            if matches:
                findings["credential_matches"][pattern_name] = {
                    "count": len(matches),
                    "samples": matches[:3],
                }

        # 计算严重程度
        cred_count = sum(
            m["count"] for m in findings["credential_matches"].values()
        )
        if findings["keyword_matches"] and cred_count > 0:
            findings["severity"] = "critical"
        elif findings["keyword_matches"]:
            findings["severity"] = "high"
        elif cred_count > 10:
            findings["severity"] = "high"
        elif cred_count > 0:
            findings["severity"] = "medium"

        return findings

    def monitor_loop(self, interval=120, iterations=None):
        """持续监控循环。"""
        count = 0
        while iterations is None or count < iterations:
            pastes = self.fetch_recent_pastes()
            alerts = []

            for paste in pastes:
                paste_key = paste.get("key", "")
                if paste_key in self.seen_keys:
                    continue
                self.seen_keys.add(paste_key)

                content = self.get_paste_content(paste_key)
                if not content:
                    continue

                findings = self.analyze_paste(content, paste)
                if findings["severity"] != "low":
                    alert = {
                        "paste_key": paste_key,
                        "title": paste.get("title", "Untitled"),
                        "user": paste.get("user", "Anonymous"),
                        "date": paste.get("date", ""),
                        "size": paste.get("size", 0),
                        "url": f"https://pastebin.com/{paste_key}",
                        "findings": findings,
                        "detected_at": datetime.now().isoformat(),
                    }
                    alerts.append(alert)
                    print(f"  [告警-{findings['severity'].upper()}] "
                          f"{paste_key}: {findings['keyword_matches']}")

            if alerts:
                self._save_alerts(alerts)

            count += 1
            if iterations is None or count < iterations:
                time.sleep(interval)

        return alerts

    def _save_alerts(self, alerts):
        """将告警保存到 JSON 文件。"""
        filename = f"{self.output_dir}/alerts_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        import os
        os.makedirs(self.output_dir, exist_ok=True)
        with open(filename, "w") as f:
            json.dump(alerts, f, indent=2)
        print(f"[+] 已保存 {len(alerts)} 条告警至 {filename}")

monitor = PastebinMonitor(
    keywords=["mycompany.com", "internal-project", "employee-name"],
)
alerts = monitor.monitor_loop(interval=120, iterations=5)

步骤 2：GitHub Gist 和代码搜索监控

class GitHubSecretMonitor:
    def __init__(self, github_token, org_keywords):
        self.token = github_token
        self.keywords = org_keywords
        self.headers = {
            "Authorization": f"token {github_token}",
            "Accept": "application/vnd.github.v3+json",
        }

    def search_code(self, query, per_page=30):
        """搜索 GitHub 代码中的泄露密钥。"""
        url = "https://api.github.com/search/code"
        params = {"q": query, "per_page": per_page}
        resp = requests.get(url, headers=self.headers, params=params)
        if resp.status_code == 200:
            results = resp.json().get("items", [])
            print(f"[+] GitHub 代码搜索：'{query}' 找到 {len(results)} 条结果")
            return results
        return []

    def search_gists(self, keyword):
        """搜索公开 Gists 中的敏感数据。"""
        url = "https://api.github.com/gists/public"
        params = {"per_page": 100}
        resp = requests.get(url, headers=self.headers, params=params)
        matches = []
        if resp.status_code == 200:
            gists = resp.json()
            for gist in gists:
                description = (gist.get("description") or "").lower()
                files = gist.get("files", {})
                for filename, file_info in files.items():
                    if keyword.lower() in description or keyword.lower() in filename.lower():
                        matches.append({
                            "gist_id": gist["id"],
                            "description": gist.get("description", ""),
                            "filename": filename,
                            "url": gist["html_url"],
                            "created_at": gist["created_at"],
                        })
        return matches

    def monitor_org_secrets(self, org_domain):
        """监控组织密钥在 GitHub 上的泄露情况。"""
        queries = [
            f'"{org_domain}" password',
            f'"{org_domain}" api_key',
            f'"{org_domain}" secret',
            f'"{org_domain}" token',
            f'"{org_domain}" credentials',
        ]
        all_findings = []
        for query in queries:
            results = self.search_code(query)
            for result in results:
                all_findings.append({
                    "query": query,
                    "repo": result.get("repository", {}).get("full_name", ""),
                    "path": result.get("path", ""),
                    "url": result.get("html_url", ""),
                    "score": result.get("score", 0),
                })
            time.sleep(10)  # GitHub 速率限制
        return all_findings

gh_monitor = GitHubSecretMonitor("YOUR_GITHUB_TOKEN", ["mycompany.com"])
findings = gh_monitor.monitor_org_secrets("mycompany.com")

步骤 3：告警与事件响应集成

def generate_credential_leak_alert(alert_data):
    """为检测到的凭证泄露生成事件告警。"""
    alert = {
        "title": f"凭证泄露检测 - {alert_data.get('severity', 'unknown').upper()}",
        "source": alert_data.get("url", ""),
        "detected_at": alert_data.get("detected_at", ""),
        "severity": alert_data.get("severity", "medium"),
        "summary": f"发现包含组织关键词和凭证的粘贴内容",
        "keyword_matches": alert_data.get("findings", {}).get("keyword_matches", []),
        "credential_types": list(alert_data.get("findings", {}).get("credential_matches", {}).keys()),
        "recommended_actions": [
            "验证泄露凭证是否有效",
            "强制受影响账户重置密码",
            "轮换暴露的 API 密钥和令牌",
            "检查访问日志是否有未授权使用记录",
            "举报粘贴内容请求下线处理",
            "如发现新模式，更新监控关键词",
        ],
    }
    return alert

验证标准

• Pastebin Scraping API 在速率限制下查询成功
• 检测到凭证模式（电子邮件:密码、API 密钥、私钥）
• 带上下文的组织特定关键词匹配
• GitHub 代码搜索识别暴露的密钥
• 生成带严重程度分类的告警
• 与事件响应工作流集成

参考资料

• Authentic8: Pastebin for CTI Research
• PasteHunter GitHub
• Scavenger Credential Crawler
• Bitdefender: Credentials on Pastebin
• zSecurity: Pastebin Monitoring
• SecurityBoulevard: Need to Monitor Paste Sites