performing-paste-site-monitoring-for-credentials

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-_]+'),
            : re.(
                ),
            : re.(
                ,
                re.IGNORECASE),
        }

     ():
        
        params = {: limit}
        :
            resp = requests.get(.SCRAPING_URL, params=params, timeout=)
             resp.status_code == :
                pastes = resp.json()
                ()
                 pastes
            :
                ()
                 []
         Exception  e:
            ()
             []

     ():
        
        params = {: paste_key}
        :
            resp = requests.get(.RAW_URL, params=params, timeout=)
             resp.status_code == :
                 resp.text
             
         Exception:
             

     ():
        
        findings = {
            : [],
            : {},
            : ,
        }

        content_lower = content.lower()

        
         keyword  .keywords:
             keyword  content_lower:
                count = content_lower.count(keyword)
                findings[].append({
                    : keyword,
                    : count,
                })

        
         pattern_name, pattern  .credential_patterns.items():
            matches = pattern.findall(content)
             matches:
                findings[][pattern_name] = {
                    : (matches),
                    : matches[:],
                }

        
        cred_count = (
            m[]  m  findings[].values()
        )
         findings[]  cred_count > :
            findings[] = 
         findings[]:
            findings[] = 
         cred_count > :
            findings[] = 
         cred_count > :
            findings[] = 

         findings

     ():
        
        count = 
         iterations    count < iterations:
            pastes = .fetch_recent_pastes()
            alerts = []

             paste  pastes:
                paste_key = paste.get(, )
                 paste_key  .seen_keys:
                    
                .seen_keys.add(paste_key)

                content = .get_paste_content(paste_key)
                  content:
                    

                findings = .analyze_paste(content, paste)
                 findings[] != :
                    alert = {
                        : paste_key,
                        : paste.get(, ),
                        : paste.get(, ),
                        : paste.get(, ),
                        : paste.get(, ),
                        : ,
                        : findings,
                        : datetime.now().isoformat(),
                    }
                    alerts.append(alert)
                    (
                          )

             alerts:
                ._save_alerts(alerts)

            count += 
             iterations    count < iterations:
                time.sleep(interval)

         alerts

     ():
        
        filename = 
         os
        os.makedirs(.output_dir, exist_ok=)
         (filename, )  f:
            json.dump(alerts, f, indent=)
        ()

monitor = PastebinMonitor(
    keywords=[, , ],
)
alerts = monitor.monitor_loop(interval=, iterations=)

步骤 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  gists:
                description = (gist.get()  ).lower()
                files = gist.get(, {})
                 filename, file_info  files.items():
                     keyword.lower()  description  keyword.lower()  filename.lower():
                        matches.append({
                            : gist[],
                            : gist.get(, ),
                            : filename,
                            : gist[],
                            : gist[],
                        })
         matches

     ():
        
        queries = [
            ,
            ,
            ,
            ,
            ,
        ]
        all_findings = []
         query  queries:
            results = .search_code(query)
             result  results:
                all_findings.append({
                    : query,
                    : result.get(, {}).get(, ),
                    : result.get(, ),
                    : result.get(, ),
                    : result.get(, ),
                })
            time.sleep()  
         all_findings

gh_monitor = GitHubSecretMonitor(, [])
findings = gh_monitor.monitor_org_secrets()

步骤 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