#!/usr/bin/env python3 import argparse import base64 import json import mmap import tempfile import os import random import re import shutil import signal import subprocess from requests.adapters import HTTPAdapter import sys import zlib from urllib3.util.retry import Retry from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED, as_completed, TimeoutError from typing import List, Dict, Any, Tuple, Optional, Union import time import threading import requests import urllib3 try: from rich.progress import ( Progress, BarColumn, TextColumn, TimeElapsedColumn, TimeRemainingColumn, MofNCompleteColumn, SpinnerColumn, ) except ImportError: Progress = None # ============================== # 1. Generic secret patterns (same style as before) # We'll treat critical/high/medium as [secret] # and low as [secret/low] # ============================== SEVERITY_PATTERNS_RAW: Dict[str, List[Tuple[str, int]]] = { "critical": [ (r"sk-[a-zA-Z0-9]{20}T3BlbkFJ[a-zA-Z0-9]{20}", 0), (r"sk-proj-[a-zA-Z0-9\-_]{80,}", 0), (r"sk-ant-api[a-zA-Z0-9\-_]{80,}", 0), (r"AKIA[A-Z0-9]{16}", 0), (r"ASIA[A-Z0-9]{16}", 0), (r"sk_live_[a-zA-Z0-9]{24,}", 0), (r"rk_live_[a-zA-Z0-9]{24,}", 0), (r"ghp_[a-zA-Z0-9]{36}", 0), (r"gho_[a-zA-Z0-9]{36}", 0), (r"github_pat_[a-zA-Z0-9_]{36,}", 0), (r"glpat-[a-zA-Z0-9\-=_]{20,}", 0), (r"-----BEGIN\s*(RSA|DSA|EC|OPENSSH|PGP)?\s*PRIVATE KEY-----", 0), (r"mongodb(\+srv)?://[^:]+:[^@]+@", 0), (r"postgres(ql)?://[^:]+:[^@]+@", 0), (r"mysql://[^:]+:[^@]+@", 0), (r"xox[baprs]-[a-zA-Z0-9\-]{10,}", 0), (r"[MN][A-Za-z0-9]{23,}\.[A-Za-z0-9_-]{6}\.[A-Za-z0-9_-]{27}", 0), (r"SK[a-f0-9]{32}", 0), (r"SG\.[a-zA-Z0-9_-]{22}\.[a-zA-Z0-9_-]{43}", 0), ], "high": [ (r"eyJ[A-Za-z0-9_-]*\.eyJ[A-Za-z0-9_-]*\.[A-Za-z0-9_-]*", 0), (r"api[_-]?key[\"'\s]*[:=][\"'\s]*[a-zA-Z0-9]{32,}", re.IGNORECASE), (r"api[_-]?secret[\"'\s]*[:=][\"'\s]*[a-zA-Z0-9]{32,}", re.IGNORECASE), (r"[Bb]earer\s+[a-zA-Z0-9\-=._+/\\]{20,500}", 0), (r"[Bb]asic\s+[A-Za-z0-9+/]{18,}={0,2}", 0), (r"AIza[0-9A-Za-z_\-]{35}", 0), (r"sk_test_[a-zA-Z0-9]{24,}", 0), (r"pk_live_[a-zA-Z0-9]{24,}", 0), (r"[0-9]+-[a-z0-9]{32}\.apps\.googleusercontent\.com", 0), (r"https://hooks\.slack\.com/services/[a-zA-Z0-9/_-]+", 0), (r"https://discord(app)?\.com/api/webhooks/\d+/[a-zA-Z0-9_-]+", 0), (r"https://oapi\.dingtalk\.com/robot/send\?access_token=[a-z0-9]+", 0), (r"sbp_[a-f0-9]{40}", 0), (r"hf_[a-zA-Z0-9]{30,}", 0), (r"shpat_[a-fA-F0-9]{32}", 0), (r"shpss_[a-fA-F0-9]{32}", 0), ], "medium": [ (r"[\"']?password[\"']?\s*[:=]\s*[\"'][^\"']{8,}[\"']", re.IGNORECASE), (r"[\"']?secret[\"']?\s*[:=]\s*[\"'][^\"']{8,}[\"']", re.IGNORECASE), (r"[\"']?token[\"']?\s*[:=]\s*[\"'][^\"']{16,}[\"']", re.IGNORECASE), (r"pk_test_[a-zA-Z0-9]{24,}", 0), (r"[\"'](10\.\d{1,3}\.\d{1,3}\.\d{1,3})[\"']", 0), (r"[\"'](172\.(1[6-9]|2\d|3[01])\.\d{1,3}\.\d{1,3})[\"']", 0), (r"[\"'](192\.168\.\d{1,3}\.\d{1,3})[\"']", 0), (r"[\"']/admin[/a-zA-Z0-9_-]*[\"']", 0), (r"[\"']/debug[/a-zA-Z0-9_-]*[\"']", 0), (r"[\"']/api/internal[/a-zA-Z0-9_-]*[\"']", 0), (r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", 0), ], "low": [ (r"[\"']?[a-z_]*key[\"']?\s*[:=]\s*[a-z_]", re.IGNORECASE), (r"[\"']?[a-z_]*token[\"']?\s*[:=]\s*[a-z_]", re.IGNORECASE), (r"[\"']?[a-z_]*secret[\"']?\s*[:=]\s*[a-z_]", re.IGNORECASE), (r"[\"']?password[\"']?\s*[:=]\s*[a-z_]", re.IGNORECASE), (r"[\"']?username[\"']?\s*[:=]\s*[a-z_]", re.IGNORECASE), (r"=\s*function", 0), (r":\s*function", 0), ], } COMPILED_SEVERITY: Dict[str, List[Tuple[re.Pattern, str]]] = { sev: [(re.compile(p, flags), p) for (p, flags) in rules] for sev, rules in SEVERITY_PATTERNS_RAW.items() } # ============================== # 2. False positive filters (from the extension) # ============================== FALSE_POSITIVE_PATTERNS_RAW: List[Tuple[str, int]] = [ (r"/cdn-cgi/challenge-platform/", 0), (r"\d+\.\d+:\d+:[a-zA-Z0-9_-]+ivzz", 0), (r"[\"']secret:[a-z]+_[a-z]+_[a-z]+[\"']", 0), (r"[a-zA-Z]+Key\s*=\s*[a-z]\.define\(", 0), (r"[a-zA-Z]+Key\s*=\s*function", 0), (r"[a-zA-Z]+Secret\s*=\s*function", 0), (r"PrivateKey\s*=\s*[a-z]\.define\(", 0), (r"PublicKey\s*=\s*[a-z]\.define\(", 0), ( r"[\"']?password[\"']?\s*[:=]\s*[\"']?(test|demo|example|placeholder|changeme|password|123456)[\"']?", re.IGNORECASE, ), ( r"[\"']?secret[\"']?\s*[:=]\s*[\"']?(test|demo|example|placeholder)[\"']?", re.IGNORECASE, ), (r"tokenize\s*:\s*function", 0), (r"getToken\s*:\s*[a-z]$", 0), (r"setPassword\s*:\s*function", 0), # HTML head/meta blocks that aren't secrets (e.g., BMW homepage metadata) (r"]+>", re.IGNORECASE), (r"]+bmw\.com", re.IGNORECASE), (r"BMWTypeNextLatin\.woff2", re.IGNORECASE), (r"The international BMW Website", re.IGNORECASE), (r"og:image.*BMW_OG\.jpg", re.IGNORECASE), ] COMPILED_FP: List[re.Pattern] = [ re.compile(p, flags) for (p, flags) in FALSE_POSITIVE_PATTERNS_RAW ] # ============================== # 3. Embedded nuclei_regex from background.js (compressed) # ============================== NUCLEI_PATTERNS_B64 = """ eJzFXQ1z2ziS/Su6uY+NJ0rGVsbZTKpSWU+Snc0mk0klubnaszwuiIRIWBRBA6RlOc5/P4AESJCS7cTz2peq2LJEvQc0Go3uxtfh4XeH0+/+cvT8QhZieXj84Og5iyJZ5WX9esHXzceHf0ynH6dTNZ3mR98fPnt61H+jeWY6XT04ut/88d14NDkaj1r4tMoEy5MaNUnrX6Vc8BwEz3Mtc1auC14wrVdSxTDgmOtFU17FzoRuaiDKtJphKNaNqDVXZ1w1LVAIrPQdRcFUmTsOxeeKa3hDbGGKMsFzX8dI8RLKRItO1RJhH6NtiZCJSlamX+u0yon69jZ4bCdfrVYNuJRJxk2bN508kksQfuGE7lq5UOKMlbzVLPt7xjR//CMNXyTznEeg1vbgihdSEfUSz1EYEqYaQcVCYcHjWf2rMoaXBBiso0baCQnuSTPsF2lR5aIMOkLzkstIxpyonU01TBtrTdQg2+DB4hN5xFVJghwnvAEuue75CGj8+UAy9o9Hd8IyIWdB4xs7LaLNiuzfFRFqiLiRCK8AVxDhdWALEYaClSbKoAWO+ZmIOC2H8QeMGdGsFES1WRjoJdGYcaYjDvQuTURXscx5NJFMzBDopYIvulCVLmXp+Aw+ELoIA9NGk4pMrntOpm3nu+KjoplVIotbD5SOuRlw5zGFIRmAE8Vl1pFyOQsYXqO4WvNSh+V1kaZ3tWCtsI1ywGN/C1C7GLo2cAIB1nicOQdur/kza1SZLdmFzNkKHG5WasZyvDa1qEtTG9csNAQ49THWfO09Z2HMLhgVaxXKlSiNYCWryrRRcuA4twmObDuH7r0LXS0JFORKEpi+GIacwvcyuJmQ6MRYgKpRFjDANCKei6RSnZtIwpKyzUSl97xsvGu9YTRphw/VziE8Ti8VP68R5aJkQ5cBqVGehwq+nUChQTXtWbl0Ic8bJ+uMKZD7E/AUVeZDhtPKJmfQBPgsegDOJ5wQ3boxdOgzxfII5JwFsL7H0hQ68FiRBDVUI3Kn9CfsjIHAU64aSyP1Vrcbw8JbT9JgaoqR16dOS1w39ZBBUIpsVq4jomFEn3XpHhBipRK4MelAM5kIFGipROF8u2qWiQgP62b1cLjDCBunCC16LQudslnG0RRSLQtWphuqjHa8rmdCuay6NI5E6qRmgyi41vcIAmcY5nVbhs4W3kH2ZpMwk1ETThDhz6pogdOrIbrP1BBmoAypMXnNRK9dElC/qJTzNqvcjA0FysToZVkEkR4KNQUOLyqKsv6aKGSPO6384gLFWewzFvGizS5rYWwLRapHn7r0C66vnepGLXGIZkBzmrhkIqOQQiFLMV/32pcoF76NisYt1wUzsHdiXjsmSotkBqI4ymQVU2hAH5yq8UmWhnrUQIFqYw1bvnIFAWEtiKEJYHN+XlHEjT2SpOhWXhWpssMyHYnvvzlbAmkU3My1oIWSJ3Z1Ida2tfBSJSwXF10WGsgRKebyqWjt6ZBZUcBln68XZJgUjlYu5UqxwvlVJMudexzg1gyhqYaozC8Ibqe8ee11odCdxGN+xjMTQjhf13oQ4E51LRWGQiS590zbhALM3dkGjmzoLfhgdR0wYEB5iQ5YeBlrGlB0ILTpzsOxSeIDN3u6kVBCqrMxhgq89CrAJCoqz+NCihwPHPM5q7LSOy5wfJLEo5HGyk5lwNUvjxMl4nZUA3q0Q2gCWLBRDqFpRIyfLvLoILSsXUfCM56LqpknTiVqYi5gyGQz/mXWB/HeCIolWfLBOhWgyK29w4kcPLhaOPjg2htU27w3GvwnOOITOOJf4YiP4YigjS4BImhHS4AI2roSIIL2qASIe3t4yF08JBwRVcYYn9GKVFWKXFxwPDSrOj8c7+ezPJ7J894UJfXUwxZKyjmIkI5m1oZlXM+lcs00qzKXYQumyivl133fJTHYD2XdbIv1XvCgw+aBh3HXkWEYHvUTcdDSP6JLJWxAu00zVOgusW18bSyHtx4ExW+hKcuOxOPnpQ2cM+fX1GWn2TXkCN0KmnYGiqghttJQNQrlyLSVAQKtqtk64UtNlvhRZWNF7QI69CStx8bvPFWyKvk+ecMqWejSz7gPzmvBEIh8wWdrQoJ22Ro6S2GXRiH1sD5Jxk9ZrPgsZs3eDbvujZahWKHUhWfcL9eDiqaDxe8JUTwROki1270zeGBg16eZOlY8FrpzKoFaF/t9SWWV5y4RaeBR4jBRkVpWWkSmh8M3jCo2m4lyeUoQhJxWAr4m4lQIt+ACCFqsnem09ZcrVP2LSqfyDJ9mcIv54UFLiAsbREJQN45jcIPlSMiFzQY3JovdWnBsR2thCR2l0HcJlx6ga7KxFhk7pViYwYUrY7JpsLtTS2pPHgMrdZmYAfE0A4+zA+B4hoVtAlu3P7f3JlhnbuSD1Syr7EY1gipksszaVes4c5N1K8fJYMEWsun9uNliBxizkuEmHAuu7MrVDD8Eh8A4oXIVwTPaHagfijGwbF24+hOtwfRdF3lmUsGUC99OwOmhgiVcxVVJYB1YtDDgXUQEVA6ptUoJhtgOF6kRHSrWrPdx0ZgnQlEsapcUY7QcpDdhwXd7Wm4QDiSKxW6f851sIvFlCHjD4txNGQqer1LhjrCH2QgL2yTTCLBzbgMJ1rn5fR0hIoK7LHIp/HlyBJgmfqZAJcDEQfrTj+pTzCaBLwA1+ley4McX2uOc5NzBQ8vsQW2SF7WTvwXF6UpEtIIiAPYTlmBXZpMAXQO06zHrxrR2LtFoXEyJjzzqYUDAEh5qeD0845p3Oxe4iSnmi/NqOTNhqLMoqndONYohQWdWO0hkiNRDBRe4gJ+f5iEpMtceG6u/HhW4B9JDkvQMj43fOzuExjWcXMHL6owlFDL2gZObBV+Hx2KBNyNcyUZz1NuVdG21UJM0LZNNg2MQEyXhm9A7UJJeakNssJ0iitpzvnIeSOZusphx/PR2zstseCgOri+xUpxxfmajMwzicq1PM2QoUwM2p5njNhHWoARhQR8XjamkpAiWOgIqXDshRCFk7LzQsredGWqIl1VWNl3XiHfRnegAOxGhTzDPnN9AgG+XYbmr6ogYiOFncgaGFrk/4Z3Ad+/QKdYvL4XLeV9wJcHQFJMyy/YAUVw5E5+q9cM4fJh1FHhcVszkOXxVYQPbFPlutqptYSRcDRWy4YPBEJ1k69OS5WK+kd5EzcltRceWPlYiI5rOMegs4XbvP8Vy2w6dQCo+yQh1njtYoIhFllT5hlnAN2XHU7i4nhQcOqiE+EqcEZYeO74GyFTyINEUN/+Ml0aUimWBL28HSyIOEmEkxvBJYuRukYkLzEeEHNhqZLIshTsfsMgkc8fl+puH24Pqejv57pyejJRw13TAkhqXhC7Y2koE76CZNI7D4Nh6ZTcfuWvTaNYxOJdLp8wvoMdVKOtCKA1eI5aJvDp3ATvBmYmZ3eEZu3XhG5KXarSp4Xh9EElaptJfbInH5wtTkyZ7093Cis18bucAGxvO8iJzR5LBpLM4l1mp80dnMoll8dMEtFV/Uc04cmtTjTdPQkWs71PBgMszRriN2t4bRbB/fcHmC7/MyZlRmNnpsEWuS5a7hC2uw3QELF46+wMr/ckKP36cxLPoaZN7h+G1swNgSD8xABOo6C4Ug095Cre+OpaRHhpQJE0enFLSKAaoH1rkJLiCNFwah8rZXcMBHi1FHrsLNnGdXZw4M+VcSuzYO0THjrrt1fHngk0o8OWSz5SbSw/WcBCkZlMZGTUhSGqmXMlFRYgKXGMVoML7Tto0XSTjetH+RuBJxxJe/wFiYeCcowekWDaTKNNzWtOCwnTbR2gWVoUbfci2pF9Bgp5RCjcPEZ3vcAUFTDt9B+qnVUB9qe2dFNEF1c0/Hleucq5KVaG81OASIZwPQHczUSJl0m6M6zef/X1PxDvP4UxLVhDtUQ9IiBJt2xlgpmYrOtA/CAjCCynqTZAggoRiOaYZpTI3wlJczzMcBFGWkWZs7VDrqAAPSnKSWsAAO/QtxETrmwcm2cmzgY9GxZnUDjOtcn9ZPDbG3c5BI+4m/0KxzGODpKnQzOd/wCFMQEc11gXxOm0fIOkCBDmGLegUHY1maZkPIgkgoaaAYHB3KYXlUpTeuuCL3McnqIDtJZKq9AE4suh1/p+mv6R4z8pNb9tJoZlzk7vhn8rK4g1r2k8Z46FpfKIhNq3Ac35e/v819zeww4LLAfNdVRVqTohsiZucQx5pkEQUqxaT7lCw8AQAoCS6i+aRh8cFuEghz0uKA0BCVCxisQLjQZXLo2bSHqYJhcStA5hLrRmNGZhnIlqoHjbSHm6BBxaduVPAKJZOzoXic7fvBSoRgzsjyDX1cMMT39yV2AQsUFm3qAQx7Fb8E406SGruNx1iTRM/pzB4vFLcL9RaafBtnf7A494pg+DyewraHXwDGigmvJCUSyA8B3V6mOfBzcoUahOpddEutwKDm5GoFJHmTEUUOzEcvgtc2ovoYBnE2PZXFY4cMPUJoIEuVqxkkbne705Oxkvdkvidr53Kz0wjo1zkWJrCMn8fBUGniu0CMWX6LVgyHpZC6jU2jSCa4oaH8cMW1IYEVPKogWlg4aA4A9JhEimcMEMby2TE3ZYBHSxFapJRKPHcxDSTMUpqQyqa+ZWYaq3SYF8PHBQpgw4VeZ0Z2a6jOE6ioUeHFEfcaES7oJ13+3dBmaArKDDY3YobXHItAIUCwgRKYldn4CyYA8QekhX7tdf1GU2ws6VtIWPpj8kB77fvg4PjvVa8BN1gCE0Aa+/cpIDFdo0Qmki+wH7H8sR5YLRLTaJ1obyTonhkxQ1V7cjefyjc5TU6YvCDY2sClmVwP6sPTDHL3WegaFwZnrwF8qiN2TbF1t02OqzB2oDHmgEDX/K8nFeNe3Q2IfTUB1yt6+Svi4mW8JNeB5RF2gyDw0OnfEnomAeMXrk3BF1bYw7yqW5ZBgru4V5TumO/BsStUtG3MZkgdbXkCjgENYDogS3n/lYX5ClXkYz9DjOoUGMzuJ+RwGZi6Tf00IySMRvcMwhniCQrZOxUWVU5/MDzqyiAUXS0eZwewRxdtJHaQu+DjLKql6jbtmHMetyPf8TydRvrUfnpetpI5EytW9y6YiVLbBen4sDhzjPmDlxB6mkfGX/b21UEhHVAJwRqfNaulewW4GETMT0aY1xtn3bnelESYR36PrS6gwqYsjtvgpCkPTIID01YamaMUUnfAnbOX5zR8ZD16hOmKKbNiVZWG9jGck44WcCaMbGETz0HqLoqkLd+BMioS+VDSKwqiy7rixawaIcm+EYnh20v5dOlQk9IOvQwlMA5Te56C6ppxChVcukjrbniGr54JWCgsikpNyqTw6eEozmFwLkqxVxEPv5Ao+d6rfESZmXpz6shWKQdwNfnIfe9ldbjhnNRBLNMJfh8BTMq3kif7jTj2TlB6t2BYpV8VuVxxrW4cDffkSyAmFUiayQ8U0YN06EZxsm93lTE/U3ld3OzyXZSwhTMTMmV0SzTpaMFXa0aEp8SIWViVoCDiTySFdWzrOJLQdI5A2TYwSY9UGy3HyB7p4sKHo6KG8R6qOiYcibyUrE1Gg/eLwJYkrKGiyKBWhwwICd5BrhgsTRz2sj9rw4SvqoKvxvKIN7NREhHRDgQGxL4FcGd03InzsuQjlZaQY1AY8IWH49ANnfUBLgg02PCC+nuO+jKC776ewsPWRe7A12n1BeHjcEMDhWiuul+42yhmiWOQZlDC79LmRrrCLxoGiI0B9q5My6XsOkflwpibfPiWtam4VhUEtixHjJRgZGuWB/5rkaQraSU5s0RSgWOGWzGD75kzfjFLkspYoIoxMATGAL8DQYesikmKO/hQW3eRoVLv7Bq4VhcT63bkCvQNU7GEiarYY+Byl349T/4MZXggJqwkDU2zGoMgYkKjQHsFudjAXElzLXwR8adsSqjWL5jb3KWzmoSX17Ecmac/hi+hIEt2YW8m5xDxxQYKlyOmWUiPPeNIOANCUgklAnWLbBBYSbSwDqhtMeNeCOzR85CiY83awHPpj2uecl5lhFq5L6CBA/e3t9HW4cNmsldkKC6iFCHD46N7a4PMrWvbJvY312bWNT61cMGYwgRX3C1IGrPtt7Alcbh9Rk4P4nGDe1QoY7XZpQJhq8FDAHD+kbYCWT7mRHe90Eh3Rtg/K7YNPhBQpiGwKsXDbpPBtGgu/6MBTcf/Pzql9fvRuah+6P3H17/fvDp1ejNq3/ZTwYPpxVbcTGdPpRam5/32OJSLy6bSpu/rWJc8jwupMjLy0xGLDPv2luId0BdkPZ+qdAe36srY0j+PPjbTwevDw8e/C97cDGdxkef9ybjR7tfzMe79ccHb16/7H38aPxj8PE/X744Ptx98JN54ujzZL/3WUN07/nTg0efLIR57OjS4B1cHvzy3vx4/dL8+PCbfWX/fFf/+N38+Pj6YMd/4fPe4y9tiRtci/jm7ScD9en9zqEpmH/00f54v1f0t5+6j81gbbDGkyfBA68vmCu8eex4On1gMYIK/Dw7qs9/m071/Q7JPPfs4fH9H6bmn6n0riHd7X+JaRHV32mQzXfu/2DYn4y/PPu8O570RfSX6fTQNNEBO6onEcQF8weTmE+Ojr43QN8fPn12VP8273139Hw6nRkhHH4qj2yfs0xGD8LyHVs+VzL3nU58SWaU/kG/Psf145PJl53uOUORpMVlkkrzvzL/tfmvLpt1dMcGY+e4AzEAj4x09/cDhIP3RnfC9pmMfxxU/uje6vywEZER0P5478mXHe899Z5afc1TSXrcPbY33ns0fMxUqfaVjuuh6NKPR5f2W6Z84739L8ftKHVphxP3pHu5ii/991fxjlWA54dWur1G6t52hfm3v/37f/znf31/9Hl/PNn90n3c9sG0LAv9dDo1KvXD6dq4h8YwGVN2LkzbPjw9NT8iuTSfRYn55MHMvv3Dis9SKRfmlTZGzUAaa/Cs16h1h2w6xCaNbFjsjt+SZYuWQsmZLDvQZsQ7rm37s1a2+7vjJ1fhFlYhH8650GllUfP2zQeG0hjmHxqCs4n54WrggG8qs33aWnadsagrsdukph1O1weOPj8e702+fMP7RrUmP3bcfP1P9fpEis2uZQSwZ7rWoDcnWXTcdXn/rNH6ybZnNQseto99sX//nT2Y138Hdmrzxejg9eijr3c+Mv/MR3uPG+Rx/Zd78DcjevOw6YqjN3z9Vc++bzYofN13DvIyVbIQ0cbjrp5ifIt3Rr9zFfHsNt/8WBX1RhZIOcJ3fPXFeBtdXX3TBUfGLtqbyofF+rs7lBperCsF8T5jOS+1cXP4ljpc99aVkC8yE8LeSmjjDZhR4+tamQ2/v+Vhd2msDb9737gG3x5D8W0k9TeuYvL6bmfKQQ2khF6y23zzgwnyV+zmrjb6YKy4cV9upBhvfukmhX7RbkTGSOMDr4ec6wtmH7mhYFug/7vQJdPpbUr1jst89PLnmx/8VCktb8PwVuTGx2zfnWypd/OIr/dGEaX1F/98r2xwRsHxMaN62L+2SdyX6gev0pQDoeq82Teqof/aqODKHiudjRpnZGupRu+lLpfs+sK6Z64S5EtpJ9M3est4yzNdmW4Wkf+K81O2F/8fRj9FJFUx+t1Od13fUvUj1+Jte37Gyijdzv6rzBMZaPmT+m3rIP3x9Oj+08M//nZ0/282otTWm+7i7k3HxB+3fO8023nu3LZvBlmu9Wl22y8rHgt9T38ze/097b8V87kV2rcQuyj9trPW/Tjpy9YUwJ9Z/fU1+OEl8yQEg3P5vy0Bfj3BoB1MY7ZHCiLPuw1gXQT2p4F/fXcUZlgmj8YmNnnYPPz5cfB68tctMUEYILnS3WNFsfO8DZDqSM/Hi7p+Mr5ve0ZTnBtVuuSZHRGatXUmeMPJ1BTk8xMTSX15GiQSBkmYzRc6NQONDfTnJlzysdPWxyL2VY8VX/eYiYNvfGwguiJjboUBwXV3HThmVqJTqflxr7Pt9tJ6gyqmVWLPFJqzCHjJUWeVn/SL8nh7UYYaKxNephx+E1U3vupFr1z7u+PHX1GuSJpCwQ/uGeqc8RbtEXHENCvOzoRPd9PRnMaKDU5RxJMsRXZWAWclA9ed5cnHpSi/LfjY6vW1UFf5r79ImRh3uQ7MRvdYHAs70LJsBxOgHVxUihPgvmzOt75ddLYyQWEmom8MLdrvXSXKt6zKo/QlU4vsuih7e0oo/PLo48s32whetOdZ3irHxXJR3ibTNvpYKusO3KaZhOJRWembn/zFUKQvfv3YvdV0j8S+Hy3hs//pukYm6L7tvJ5euvuY0OsXOio7g3bYzqVY/7ae6Hvz6l+XH1+9+PDq0+Wn3968enf5/uDjx//57cPLS/Pey1fvPr0+eHtptHinR/msXwAfvhhnq8luH/0f7Qin0A== """.strip() # ============================== # 4. Core scanning helpers (multi-threaded) # ============================== REGEX_CHUNK_SIZE = 50 LABEL_ORDER = {"secret": 0, "secret/low": 1} DEFAULT_CONNECT_TIMEOUT = 5 DEFAULT_READ_TIMEOUT = 90 DEFAULT_USE_CURL = True # prefer curl path that worked without flags; may disable for high parallelism DEFAULT_SCAN_TIMEOUT = 80.0 DEFAULT_FETCH_TIMEOUT = 30.0 # wall-clock timeout per fetch batch DEFAULT_REQUIRE_STATUS_200 = False DEFAULT_MAX_BODY_BYTES = None # No limit - stream everything to disk for complete scanning MAX_PARALLEL_HARD_LIMIT = 50 # hard cap on simultaneous fetch threads PROGRESS_BAR_WIDTH = 20 # Chunked scanning for gigabyte files SCAN_CHUNK_SIZE = 50 * 1024 * 1024 # 50MB per chunk OVERLAP_SIZE = 10 * 1024 # 10KB overlap for boundary matches LARGE_FILE_THRESHOLD = 50 * 1024 * 1024 # Files > 50MB use chunked scanning # Parallel pattern processing PATTERN_WORKERS = min(os.cpu_count() or 4, 16) # Scale with CPU cores FAILED_FETCH_LOG = os.path.join(tempfile.gettempdir(), "metho_failed_to_fetch.txt") ANSI_RESET = "\033[0m" ANSI_COLORS_LABEL = { "secret": "\033[93m", # yellow "secret/low": "\033[90m", # gray } ANSI_MATCH = "\033[96m" # cyan ANSI_SOURCE = "\033[95m" # magenta STOP_EVENT = threading.Event() QUIET = False thread_local = threading.local() RG_PATH = shutil.which("rg") RG_SEM = threading.Semaphore(8) if RG_PATH else None RG_ENABLED_LOCK = threading.Lock() RG_ENABLED = bool(RG_PATH) # Proxy management PROXY_LIST: List[str] = [] PROXY_LOCK = threading.Lock() def parse_proxy_string(proxy_raw: str) -> Optional[str]: """ Parse a proxy string in format ip:port:username:password and return a URL in format http://username:password@ip:port Returns None if parsing fails. """ proxy_raw = proxy_raw.strip() if not proxy_raw: return None parts = proxy_raw.split(":") if len(parts) == 4: ip, port, username, password = parts return f"http://{username}:{password}@{ip}:{port}" elif len(parts) == 2: # Simple ip:port format without auth ip, port = parts return f"http://{ip}:{port}" else: debug_log(f"[!] Invalid proxy format: {proxy_raw} (expected ip:port:user:pass or ip:port)\n") return None def load_proxies(proxy_input: str) -> List[str]: """ Load proxies from either: - A comma-separated string of proxies - A file path containing one proxy per line Proxy format: ip:port:username:password Returns list of proxy URLs in format http://username:password@ip:port """ proxies: List[str] = [] # Check if it's a file path if os.path.isfile(proxy_input): try: with open(proxy_input, "r", encoding="utf-8", errors="ignore") as f: for line in f: line = line.strip() if line and not line.startswith("#"): proxy_url = parse_proxy_string(line) if proxy_url: proxies.append(proxy_url) except OSError as e: debug_log(f"[!] Failed to read proxy file {proxy_input}: {e}\n") else: # Treat as comma-separated string for proxy_raw in proxy_input.split(","): proxy_url = parse_proxy_string(proxy_raw) if proxy_url: proxies.append(proxy_url) return proxies def get_random_proxy() -> Optional[str]: """Get a random proxy from the proxy list. Returns None if no proxies configured.""" with PROXY_LOCK: if not PROXY_LIST: return None return random.choice(PROXY_LIST) # Ensure Ctrl+C stops all work immediately def _sigint_handler(signum, frame): STOP_EVENT.set() debug_log("[!] Interrupt received, stopping...\n") raise KeyboardInterrupt signal.signal(signal.SIGINT, _sigint_handler) def debug_log(msg: str) -> None: if QUIET: return try: sys.stderr.write(msg) sys.stderr.flush() except Exception: pass def is_false_positive(context: str, use_fp: bool) -> bool: if not use_fp: return False return any(fp.search(context) for fp in COMPILED_FP) def load_embedded_nuclei_patterns() -> List[Tuple[re.Pattern, str]]: raw_bytes = zlib.decompress(base64.b64decode(NUCLEI_PATTERNS_B64)) raw_patterns: List[List[Any]] = json.loads(raw_bytes.decode("utf-8")) compiled: List[Tuple[re.Pattern, str]] = [] for pattern, flags in raw_patterns: pattern_str = str(pattern).strip() if not pattern_str: continue try: compiled.append((re.compile(pattern_str, flags), pattern_str)) except re.error as e: # Skip malformed ones so the scanner keeps running debug_log(f"[!] Failed to compile nuclei regex: /{pattern_str}/ : {e}\n") return compiled def filter_nuclei_patterns( patterns: List[Tuple[re.Pattern, str]], keep_all: bool ) -> List[Tuple[re.Pattern, str]]: if keep_all: return patterns keywords = [ "key", "secret", "token", "bearer", "api", "auth", "password", "passphrase", "private", "credential", "aws", "github", "gitlab", "slack", "discord", "webhook", "xox", "sg.", "sk_", "pk_", "rk_", "ssh", "jwt", ] filtered: List[Tuple[re.Pattern, str]] = [] for rx, pat in patterns: low = pat.lower() if any(k in low for k in keywords): filtered.append((rx, pat)) return filtered def chunk_patterns( patterns: List[Tuple[re.Pattern, str]], chunk_size: int ) -> List[List[Tuple[re.Pattern, str]]]: return [patterns[i : i + chunk_size] for i in range(0, len(patterns), chunk_size)] def deduplicate_findings(findings: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Remove duplicate findings while preserving order.""" seen = set() unique: List[Dict[str, Any]] = [] for f in findings: key = (f.get("severity"), f.get("match"), f.get("source")) if key in seen: continue seen.add(key) unique.append(f) return unique def inline_flags_pattern(pattern: str, flags: int) -> str: """ Prepend inline PCRE flags for rg based on Python regex flags. Only adds the commonly used flags we support. """ prefix = "" if flags & re.IGNORECASE: prefix += "(?i)" if flags & re.MULTILINE: prefix += "(?m)" if flags & re.DOTALL: prefix += "(?s)" return prefix + pattern if prefix else pattern def ensure_rg_pcre2_available() -> None: """Fail fast if rg is present but lacks PCRE2 support.""" if not RG_PATH: return cmd = [RG_PATH, "-P", "--version"] try: proc = subprocess.run( cmd, capture_output=True, text=True, encoding="utf-8", errors="ignore", ) except Exception as e: debug_log(f"[!] Failed to run rg for PCRE2 check: {e}\n") return if proc.returncode != 0: msg = (proc.stderr or proc.stdout or "").strip() raise SystemExit( "[!] ripgrep is installed without PCRE2 support. Install a PCRE2-enabled ripgrep " "(e.g., package 'ripgrep-pcre2' or 'cargo install ripgrep --features \"pcre2\" --locked').\n" f"rg output: {msg}" ) def scan_one_chunk( label: str, chunk: List[Tuple[re.Pattern, str]], text: str, source: str, use_fp: bool, tmp_path: Optional[str], use_rg: bool, ) -> List[Dict[str, Any]]: """Scan a chunk of regex patterns sequentially.""" global RG_ENABLED results: List[Dict[str, Any]] = [] with RG_ENABLED_LOCK: rg_allowed = bool(use_rg and tmp_path and RG_PATH and RG_ENABLED) if rg_allowed: for rx, pattern_str in chunk: cmd = [ RG_PATH, "--pcre2", "--multiline", "--text", "--json", "--byte-offset", "--mmap", "--regex-size-limit", "100M", "--dfa-size-limit", "100M", "-o", "-e", pattern_str, tmp_path, ] try: if RG_SEM: RG_SEM.acquire() proc = subprocess.run( cmd, capture_output=True, text=True, encoding="utf-8", errors="ignore", ) except Exception as e: debug_log(f"[!] rg failed for {source}: {e}\n") with RG_ENABLED_LOCK: RG_ENABLED = False rg_allowed = False break finally: if RG_SEM: RG_SEM.release() if proc.returncode not in (0, 1): # 1 = no matches stderr_msg = proc.stderr.strip() if proc.returncode == 2 and proc.stdout.strip(): # Got partial matches despite errors - continue processing debug_log(f"[!] rg exit {proc.returncode} for {source}: {stderr_msg} (partial results)\n") elif "match limit exceeded" in stderr_msg: # Skip this pattern but continue with others debug_log(f"[!] rg match limit for {source}: {stderr_msg} (skipping pattern)\n") continue else: debug_log(f"[!] rg exit {proc.returncode} for {source}: {stderr_msg}\n") with RG_ENABLED_LOCK: RG_ENABLED = False rg_allowed = False break for line in proc.stdout.splitlines(): try: rec = json.loads(line) except json.JSONDecodeError: continue if rec.get("type") != "match": continue for sub in rec.get("data", {}).get("submatches", []): if "match" not in sub or "byte_offset" not in sub: continue start = sub["byte_offset"] match_text = sub["match"]["text"] ctx_start = max(0, start - 60) ctx_end = min(len(text), start + len(match_text) + 60) context = text[ctx_start:ctx_end].replace("\n", " ") if is_false_positive(context, use_fp): continue results.append( { "severity": label, "source": source, "index": start, "match": match_text, "pattern": pattern_str, "context": context.strip(), } ) if not rg_allowed: for rx, pattern_str in chunk: for m in rx.finditer(text): start, end = m.start(), m.end() ctx_start = max(0, start - 60) ctx_end = min(len(text), end + 60) context = text[ctx_start:ctx_end].replace("\n", " ") if is_false_positive(context, use_fp): continue results.append( { "severity": label, # "secret" or "secret/low" "source": source, "index": start, "match": m.group(0), "pattern": pattern_str, "context": context.strip(), } ) return results def scan_text_grouped( text: str, source: str, pattern_groups: Dict[str, List[Tuple[re.Pattern, str]]], use_fp: bool, chunk_size: int = REGEX_CHUNK_SIZE, ) -> Tuple[List[Dict[str, Any]], int]: # Deprecated in favor of explicit chunk submission in main loop raise NotImplementedError("scan_text_grouped is not used in the current pipeline") # ============================== # 5. Fetching, formatting, CLI # ============================== def build_session( retries: int, backoff: float, trust_env: bool, pool_maxsize: Optional[int] = None ) -> requests.Session: """Create a requests session with retry/backoff to survive flaky endpoints.""" session = requests.Session() session.trust_env = trust_env retry_cfg = Retry( total=retries, connect=retries, read=retries, status=retries, backoff_factor=backoff, allowed_methods=["GET", "HEAD", "OPTIONS"], status_forcelist=[429, 500, 502, 503, 504], raise_on_status=False, ) adapter_kwargs = {"max_retries": retry_cfg} if pool_maxsize: adapter_kwargs["pool_maxsize"] = pool_maxsize adapter_kwargs["pool_connections"] = pool_maxsize adapter = HTTPAdapter(**adapter_kwargs) session.mount("http://", adapter) session.mount("https://", adapter) return session def get_thread_session( retries: int, backoff: float, trust_env: bool, pool_maxsize: Optional[int], ) -> requests.Session: """Provide one session per thread to avoid cross-thread contention.""" sess = getattr(thread_local, "session", None) if sess is None: sess = build_session(retries, backoff, trust_env=trust_env, pool_maxsize=pool_maxsize) thread_local.session = sess return sess def fetch_with_curl( url: str, connect_timeout: float, read_timeout: float, verify: bool, user_agent: Optional[str], use_proxy_env: bool, curl_path: Optional[str], verbose: bool, max_body_bytes: Optional[int], require_status_200: bool, dest_path: Optional[str] = None, proxy: Optional[str] = None, ) -> str: total_timeout = max(int(connect_timeout + read_timeout), int(read_timeout)) curl_bin = curl_path or shutil.which("curl") or "curl" cmd = [ curl_bin, "-L", "--connect-timeout", str(connect_timeout), "--max-time", str(total_timeout), ] if max_body_bytes is not None: cmd.extend(["--max-filesize", str(max_body_bytes)]) if not verify: cmd.append("-k") if user_agent: cmd.extend(["-A", user_agent]) if verbose: cmd.append("-v") if require_status_200: cmd.extend(["-w", "\\n%{http_code}"]) if dest_path: cmd.extend(["-o", dest_path]) if proxy: cmd.extend(["-x", proxy]) cmd.append(url) env = None if not use_proxy_env and not proxy: env = dict(os.environ) for k in list(env.keys()): if k.upper() in {"HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"}: env.pop(k, None) env["NO_PROXY"] = "*" result = subprocess.run( cmd, capture_output=True, text=True, env=env, encoding="utf-8", errors="ignore" ) if result.returncode != 0: stderr_msg = result.stderr.strip() raise RuntimeError(f"curl exited with code {result.returncode}: {stderr_msg}") stdout = result.stdout if require_status_200: stdout, _, status_str = stdout.rpartition("\n") try: status_code = int(status_str.strip()) except ValueError: raise RuntimeError( f"Could not parse HTTP status from curl output for {url}: {status_str!r}" ) if status_code != 200: raise RuntimeError(f"HTTP status {status_code} for {url}, skipping.") return stdout def fetch_url_body( url: str, session: requests.Session, connect_timeout: float, read_timeout: float, verify: bool, user_agent: Optional[str], use_curl: bool, curl_path: Optional[str], curl_verbose: bool, trust_env: bool, max_body_bytes: Optional[int], dest_path: Optional[str], require_status_200: bool, proxy: Optional[str] = None, ) -> str: if use_curl: return fetch_with_curl( url=url, connect_timeout=connect_timeout, read_timeout=read_timeout, verify=verify, user_agent=user_agent, use_proxy_env=trust_env, curl_path=curl_path, verbose=curl_verbose, max_body_bytes=max_body_bytes, require_status_200=require_status_200, dest_path=dest_path, proxy=proxy, ) headers = {} if user_agent: headers["User-Agent"] = user_agent proxies = {"http": proxy, "https": proxy} if proxy else None if dest_path: tmp_path = dest_path # Stream to file to avoid buffering huge bodies in memory try: with session.get( url, headers=headers, timeout=(connect_timeout, read_timeout), verify=verify, stream=True, proxies=proxies, ) as resp, open(tmp_path, "wb") as out: if require_status_200 and resp.status_code != 200: raise RuntimeError(f"HTTP status {resp.status_code} for {url}, skipping.") total_bytes = 0 for chunk in resp.iter_content(chunk_size=16384): if not chunk: continue total_bytes += len(chunk) if max_body_bytes is not None and total_bytes > max_body_bytes: raise RuntimeError( f"Body size exceeded limit ({max_body_bytes} bytes), skipping." ) out.write(chunk) return tmp_path except Exception: # On failure, clean partial file if present try: if os.path.exists(dest_path): os.remove(dest_path) except OSError: pass raise with session.get( url, headers=headers, timeout=(connect_timeout, read_timeout), verify=verify, stream=True, proxies=proxies, ) as resp: if require_status_200 and resp.status_code != 200: raise RuntimeError(f"HTTP status {resp.status_code} for {url}, skipping.") body_chunks: List[bytes] = [] total_bytes = 0 for chunk in resp.iter_content(chunk_size=16384, decode_unicode=False): if not chunk: continue total_bytes += len(chunk) if max_body_bytes is not None and total_bytes > max_body_bytes: raise RuntimeError( f"Body size exceeded limit ({max_body_bytes} bytes), skipping." ) body_chunks.append(chunk) encoding = resp.encoding or "utf-8" return b"".join(body_chunks).decode(encoding, errors="ignore") def normalize_url(raw: str, auto_http: bool) -> Optional[str]: """ Clean up a URL string. If the scheme is missing and auto_http is True, prepend http:// for host-like entries. Returns None for empty/invalid lines. """ url = raw.strip() if not url or url.startswith("#"): return None if url.startswith("//"): return f"http:{url}" if auto_http else None if "://" not in url: if auto_http and re.match(r"^[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}(:\d+)?(/.*)?$", url): return f"http://{url}" debug_log(f"[!] Skipping URL without scheme: {url}\n") return None return url def record_failed_urls(urls: List[str], log_path: str = FAILED_FETCH_LOG) -> None: """Append failed URLs to a log file so they can be retried later.""" if not urls: return try: with open(log_path, "a", encoding="utf-8") as f: for url in urls: if url: f.write(url.strip() + "\n") except Exception as e: debug_log(f"[!] Failed to record failed URLs to {log_path}: {e}\n") def get_state_file_path(list_file: Optional[str], output_file: Optional[str]) -> str: """Generate state file path based on output or list file.""" if output_file: return output_file + ".state" elif list_file: base = os.path.splitext(os.path.basename(list_file))[0] return f"{base}_scan.state" return ".scan_state.json" def save_scan_state( state_path: str, remaining_urls: List[str], processed_count: int, output_file: Optional[str], list_file: Optional[str], ) -> None: """Save scan progress to a state file for resume capability.""" state = { "remaining_urls": remaining_urls, "processed_count": processed_count, "output_file": output_file, "list_file": list_file, "timestamp": time.time(), } try: with open(state_path, "w", encoding="utf-8") as f: json.dump(state, f, indent=2) except Exception as e: debug_log(f"[!] Failed to save scan state: {e}\n") def load_scan_state(state_path: str) -> Optional[Dict[str, Any]]: """Load scan state from file. Returns None if file doesn't exist or is invalid.""" if not os.path.exists(state_path): return None try: with open(state_path, "r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, OSError) as e: debug_log(f"[!] Failed to load scan state from {state_path}: {e}\n") return None def clear_scan_state(state_path: str) -> None: """Remove state file after successful completion.""" try: if os.path.exists(state_path): os.remove(state_path) except OSError: pass def load_urls_from_file(path: str, auto_http: bool) -> List[str]: urls: List[str] = [] with open(path, "r", encoding="utf-8", errors="ignore") as f: for line in f: norm = normalize_url(line, auto_http=auto_http) if norm: urls.append(norm) return urls def format_finding(rec: Dict[str, Any], color: bool) -> str: """ Format as: [secret] 'match' [source] with different colors when color=True. """ label = rec["severity"] # "secret" or "secret/low" match_repr = repr(rec["match"]) src = rec["source"] if color: lab = label lab_color = ANSI_COLORS_LABEL.get(label, "") if lab_color: lab = f"{lab_color}{lab}{ANSI_RESET}" match_repr = f"{ANSI_MATCH}{match_repr}{ANSI_RESET}" src = f"{ANSI_SOURCE}{src}{ANSI_RESET}" return f"[{lab}] {match_repr} [{src}]" else: return f"[{label}] {match_repr} [{src}]" def main(): parser = argparse.ArgumentParser( description=( "Web Secret Scanner - Detect exposed secrets, API keys, tokens, and credentials in web responses.\n\n" "Scans URLs for sensitive data leaks using pattern matching with nuclei-regex signatures\n" "and custom secret detection rules. Supports single URL (-u) or batch scanning from file (-l).\n\n" "Output format: [secret] 'matched_value' [source_url]" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("-u", "--url", help="Single URL to scan.") parser.add_argument("-l", "--list", help="File containing URLs (one per line).") parser.add_argument( "--timeout", type=float, default=None, help="Request timeout in seconds (connect + read). Default: 95s total.", ) parser.add_argument( "--retries", type=int, default=3, help="Number of HTTP retries for fetch errors. Default: 3.", ) parser.add_argument( "--secure", dest="insecure", action="store_false", default=True, help="Enable TLS certificate verification. Default: disabled (insecure).", ) parser.add_argument( "--proxy", type=str, default=None, help=( "Proxy configuration: comma-separated proxies or path to file. " "Format: ip:port:username:password (e.g., 185.195.222.170:3199:user:pass). " "Proxies are randomly shuffled for each request." ), ) parser.add_argument( "--user-agent", type=str, default=None, nargs="?", const="", help="Override User-Agent header (empty to use client default).", ) parser.add_argument( "--npb", dest="progress", action="store_false", default=True, help="No Progress Bar - disable the progress bar display.", ) parser.add_argument( "--quiet", action="store_true", help="Suppress all terminal output except the progress bar.", ) parser.add_argument( "--chunk-size", type=int, default=REGEX_CHUNK_SIZE, help="Number of regex patterns per processing chunk. Default: 50.", ) parser.add_argument( "--all-patterns", action="store_true", help="Include noisy patterns (IPs, endpoints, emails). Default: secrets-only.", ) parser.add_argument( "-o", "--output", type=str, default=None, help="Output file path. Default: _secrets.txt when --list is used.", ) parser.add_argument( "--html-output", nargs="?", const="", help="Generate HTML report. Optional path (defaults to .html).", ) parser.add_argument( "--append", action="store_true", help="Append to the output file instead of overwriting.", ) parser.add_argument( "--machine-progress", action="store_true", help="Emit [progress] N/M lines to stderr for machine consumption.", ) parser.add_argument( "--resume", action="store_true", help="Resume a previous interrupted scan session from the saved state.", ) parser.add_argument( "--nou", dest="number_of_urls", type=int, default=6, help=( "Number of URLs to fetch concurrently. Default: 6. " "NOTE: This controls URL fetching parallelism only; actual thread count may be " "significantly higher due to regex processing workers." ), ) parser.add_argument( "--no-color", dest="color", action="store_false", default=True, help="Disable colored output.", ) parser.add_argument( "--scan-chunk-mb", type=int, default=50, help="Chunk size in MB for scanning large files. Default: 50MB.", ) parser.add_argument( "--pattern-workers", type=int, default=None, help=f"Parallel workers for pattern scanning. Default: {PATTERN_WORKERS} (auto-detected).", ) parser.add_argument( "--max-file-size", type=int, default=None, help="Maximum file size in MB to process. Default: unlimited.", ) parser.add_argument( "--max-blocks", type=int, default=2000, help="Max result blocks per output file before splitting into a new part. Default: 2000.", ) args = parser.parse_args() # Store effective config values in args for use throughout args._scan_chunk_size = (args.scan_chunk_mb * 1024 * 1024) if args.scan_chunk_mb else SCAN_CHUNK_SIZE args._pattern_workers = args.pattern_workers if args.pattern_workers else PATTERN_WORKERS args._large_file_threshold = args._scan_chunk_size # Files > chunk size use chunked scanning if args.output is None and args.list: list_stem = os.path.splitext(os.path.basename(args.list.rstrip(os.sep)))[0] or "scan" args.output = f"{list_stem}_secrets.txt" html_requested = args.html_output is not None html_output_path: Optional[str] = None if html_requested: if not args.output: parser.error("--html-output requires --output to be set.") base, _ = os.path.splitext(args.output) html_output_path = args.html_output if args.html_output else (base or args.output) + ".html" # Reset globals per run STOP_EVENT.clear() with RG_ENABLED_LOCK: RG_ENABLED = bool(RG_PATH) # Ensure rg supports PCRE2; fail fast with guidance if not ensure_rg_pcre2_available() if not args.url and not args.list: parser.error("You must provide --url or --list") if args.progress and Progress is None: debug_log("[i] Disabling progress (rich not installed).\n") args.progress = False # Set global quiet flag for helpers global QUIET QUIET = args.quiet # Initialize proxy list if provided global PROXY_LIST if args.proxy: PROXY_LIST = load_proxies(args.proxy) if PROXY_LIST: debug_log(f"[i] Loaded {len(PROXY_LIST)} proxies for rotation.\n") else: debug_log("[!] No valid proxies loaded from --proxy argument.\n") else: PROXY_LIST = [] max_parallel = max(1, args.number_of_urls) if max_parallel > MAX_PARALLEL_HARD_LIMIT: debug_log( f"[i] Limiting parallel URL workers to {MAX_PARALLEL_HARD_LIMIT} " f"(requested {max_parallel})." ) max_parallel = MAX_PARALLEL_HARD_LIMIT fetch_workers = max_parallel pool_maxsize = fetch_workers * 2 # allow multiple connections per worker # Automatically avoid spawning tons of curl processes at high concurrency use_curl = DEFAULT_USE_CURL if use_curl and max_parallel > 8: debug_log("[i] Disabling curl for high concurrency to reduce process overhead.") use_curl = False if args.insecure: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) nuclei_patterns = filter_nuclei_patterns( load_embedded_nuclei_patterns(), keep_all=False ) # Build pattern groups: # [secret] => nuclei_regex + critical/high/medium # [secret/low] => low-generic patterns # Build pattern groups with optional exclusion of noisy patterns noisy_medium_patterns = { r"[\"'](10\.\d{1,3}\.\d{1,3}\.\d{1,3})[\"']", r"[\"'](172\.(1[6-9]|2\d|3[01])\.\d{1,3}\.\d{1,3})[\"']", r"[\"'](192\.168\.\d{1,3}\.\d{1,3})[\"']", r"[\"']/admin[/a-zA-Z0-9_-]*[\"']", r"[\"']/debug[/a-zA-Z0-9_-]*[\"']", r"[\"']/api/internal[/a-zA-Z0-9_-]*[\"']", r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", } secret_patterns: List[Tuple[re.Pattern, str]] = [] secret_patterns.extend(nuclei_patterns) secret_patterns.extend(COMPILED_SEVERITY["critical"]) secret_patterns.extend(COMPILED_SEVERITY["high"]) for rx, pat in COMPILED_SEVERITY["medium"]: if args.all_patterns or pat not in noisy_medium_patterns: secret_patterns.append((rx, pat)) secret_low_patterns: List[Tuple[re.Pattern, str]] = COMPILED_SEVERITY["low"] pattern_groups: Dict[str, List[Tuple[re.Pattern, str]]] = { "secret": secret_patterns, "secret/low": secret_low_patterns, } # Collect URLs urls: List[str] = [] auto_http = True # Always auto-prepend http:// for URLs without scheme if args.url: norm = normalize_url(args.url, auto_http=auto_http) if norm: urls.append(norm) if args.list: urls.extend(load_urls_from_file(args.list, auto_http=auto_http)) # Handle resume logic state_file = get_state_file_path(args.list, args.output) resumed_processed = 0 if args.resume: state = load_scan_state(state_file) if state: remaining_from_state = state.get("remaining_urls", []) processed_from_state = state.get("processed_count", 0) if remaining_from_state: debug_log( f"[i] Resuming scan: {processed_from_state} URLs already processed, " f"{len(remaining_from_state)} remaining.\n" ) urls = remaining_from_state resumed_processed = processed_from_state # Ensure we append to output file when resuming args.append = True else: debug_log("[i] Previous scan appears complete. Starting fresh.\n") clear_scan_state(state_file) else: debug_log("[i] No previous scan state found. Starting fresh scan.\n") # Prepare pattern chunks for rg to avoid PCRE2 size issues max_pattern_len = 512 # tighter cap to avoid PCRE2 size errors patterns_dir = os.path.join(tempfile.gettempdir(), "metho_patterns_split") os.makedirs(patterns_dir, exist_ok=True) for entry in os.scandir(patterns_dir): try: os.remove(entry.path) except OSError: pass patterns_per_file = 50 pattern_lines: List[str] = [] skipped_patterns = 0 for rules in [ nuclei_patterns, COMPILED_SEVERITY["critical"], COMPILED_SEVERITY["high"], COMPILED_SEVERITY["medium"], COMPILED_SEVERITY["low"], ]: for rx, pat in rules: pat_inline = inline_flags_pattern(pat, rx.flags) if len(pat_inline) > max_pattern_len: skipped_patterns += 1 continue pattern_lines.append(pat_inline) pattern_files: List[str] = [] for i in range(0, len(pattern_lines), patterns_per_file): chunk_lines = pattern_lines[i : i + patterns_per_file] p_path = os.path.join(patterns_dir, f"patterns_{i//patterns_per_file}.txt") with open(p_path, "w", encoding="utf-8", errors="ignore") as pf: pf.write("\n".join(chunk_lines)) pattern_files.append(p_path) debug_log(f"[i] Wrote {len(pattern_lines)} patterns into {len(pattern_files)} chunk file(s) under {patterns_dir}") if skipped_patterns: debug_log(f"[i] Skipped {skipped_patterns} patterns that exceeded {max_pattern_len} chars (PCRE2 limit)") # Build pattern chunks for Python fallback scanning pattern_chunks: List[Tuple[str, List[Tuple[re.Pattern, str]]]] = [] for label, rules in pattern_groups.items(): if not rules: continue for chunk in chunk_patterns(rules, args.chunk_size): pattern_chunks.append((label, chunk)) temp_dir = os.path.join(tempfile.gettempdir(), "metho_findsomething_temp") temp_dir2 = os.path.join(tempfile.gettempdir(), "metho_findsomething_temp2") os.makedirs(temp_dir, exist_ok=True) os.makedirs(temp_dir2, exist_ok=True) def clear_dir(path: str): for entry in os.scandir(path): try: os.remove(entry.path) except OSError: pass def fetch_batch(batch_urls: List[str], directory: str) -> Tuple[Dict[str, str], List[str]]: os.makedirs(directory, exist_ok=True) mapping: Dict[str, str] = {} failed_urls: List[str] = [] if not batch_urls: return mapping, failed_urls batch_timeout = DEFAULT_FETCH_TIMEOUT def worker(url: str) -> Optional[Tuple[str, str]]: if STOP_EVENT.is_set(): return None session = get_thread_session(**session_params) # Get a random proxy for this request (shuffled rotation) proxy = get_random_proxy() try: fname = os.path.join(directory, f"u{hash(url)}_{int(time.time()*1000)}.txt") fetch_url_body( url, session=session, connect_timeout=connect_timeout, read_timeout=read_timeout, verify=not args.insecure, user_agent=user_agent, use_curl=use_curl, curl_path=None, curl_verbose=False, trust_env=trust_env, max_body_bytes=max_body_bytes, dest_path=fname, require_status_200=DEFAULT_REQUIRE_STATUS_200, proxy=proxy, ) except Exception as e: failed_urls.append(url) return None return fname, url with ThreadPoolExecutor(max_workers=fetch_workers) as ex: fut_map = {ex.submit(worker, u): u for u in batch_urls} try: for fut in as_completed(fut_map, timeout=batch_timeout): res = fut.result() if res: fname, url = res mapping[fname] = url except TimeoutError: for fut, url in fut_map.items(): if not fut.done(): fut.cancel() failed_urls.append(url) except KeyboardInterrupt: STOP_EVENT.set() ex.shutdown(wait=False, cancel_futures=True) raise if failed_urls: fail_log = os.path.join(os.path.dirname(os.path.abspath(args.output)), "failed_to_fetch.txt") if args.output else FAILED_FETCH_LOG record_failed_urls(sorted(set(failed_urls)), log_path=fail_log) return mapping, failed_urls def run_rg_on_dir(directory: str, file_map: Dict[str, str], pattern_files: List[str]) -> Dict[str, List[Dict[str, Any]]]: # Normalize mapping keys to absolute, case-folded paths for rg output def _norm(p: str) -> str: return os.path.normcase(os.path.abspath(p)) abs_map: Dict[str, str] = {_norm(p): u for p, u in file_map.items()} results: Dict[str, List[Dict[str, Any]]] = {url: [] for url in abs_map.values()} if not abs_map or not pattern_files: return results global RG_ENABLED with RG_ENABLED_LOCK: if not RG_ENABLED: return results rg_timeout = DEFAULT_SCAN_TIMEOUT for pf in pattern_files: if STOP_EVENT.is_set(): break cmd = [ RG_PATH or "rg", "--pcre2", "--multiline", "--text", "--json", "--byte-offset", # Performance flags for large files "--mmap", # Memory-map files for faster I/O "--regex-size-limit", "100M", # Allow larger compiled patterns "--dfa-size-limit", "100M", # Allow larger DFA for complex patterns "--no-ignore", # Skip gitignore processing "--no-config", # Skip config file loading "-f", pf, directory, ] try: proc = subprocess.run( cmd, capture_output=True, text=True, encoding="utf-8", errors="ignore", timeout=rg_timeout, ) except subprocess.TimeoutExpired: debug_log(f"[!] rg timed out after {rg_timeout}s using {pf}; skipping remaining pattern chunks.\n") break except Exception as e: debug_log(f"[!] rg failed to start: {e}\n") with RG_ENABLED_LOCK: RG_ENABLED = False break if proc.returncode not in (0, 1): # 1 = no matches stderr_msg = proc.stderr.strip() # Exit code 2 can mean partial matches with some errors - check if we got results if proc.returncode == 2 and proc.stdout.strip(): # Got some matches despite errors (e.g., match limit on some files) # Log warning but continue processing the results we did get debug_log(f"[!] rg exit {proc.returncode} on {pf}: {stderr_msg} (continuing with partial results)\n") elif "match limit exceeded" in stderr_msg: # PCRE2 match limit hit - file is too complex for this pattern # Continue to next pattern file rather than disabling rg entirely debug_log(f"[!] rg match limit exceeded on {pf}: {stderr_msg} (skipping this pattern chunk)\n") continue elif "PCRE2 is not available" in stderr_msg: raise SystemExit( "[!] ripgrep is installed without PCRE2 support. Install a PCRE2-enabled ripgrep " "(e.g., package 'ripgrep-pcre2' or 'cargo install ripgrep --features \"pcre2\" --locked')." ) else: debug_log(f"[!] rg exit {proc.returncode} on {pf}: {stderr_msg}\n") with RG_ENABLED_LOCK: RG_ENABLED = False break for line in proc.stdout.splitlines(): try: rec = json.loads(line) except json.JSONDecodeError: continue if rec.get("type") != "match": continue path = rec.get("data", {}).get("path", {}).get("text") if not path: continue abs_path = _norm(path) if abs_path not in abs_map: continue url = abs_map[abs_path] for sub in rec.get("data", {}).get("submatches", []): if "match" not in sub: continue start = sub.get("byte_offset", sub.get("start")) if start is None: continue match_text = sub["match"].get("text", "") line_text = rec.get("data", {}).get("lines", {}).get("text", "") if len(line_text) > 500: line_text = line_text[:500] + "...(truncated)" results[url].append( { "severity": "secret", "source": url, "index": start, "match": match_text, "pattern": rec.get("data", {}).get("pattern", ""), "context": line_text.strip(), } ) return results def scan_large_file_chunked( file_path: str, url: str, patterns: List[Tuple[re.Pattern, str]], label: str, use_fp: bool, ) -> List[Dict[str, Any]]: """ Scan gigabyte-sized files in overlapping chunks using mmap. Prevents OOM while ensuring no line is missed at chunk boundaries. """ results: List[Dict[str, Any]] = [] seen_matches: set = set() # Dedup boundary matches by (abs_position, match_text) try: file_size = os.path.getsize(file_path) except OSError: return results if file_size == 0: return results try: with open(file_path, "rb") as f: # Use mmap for memory-efficient access with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm: offset = 0 while offset < file_size: if STOP_EVENT.is_set(): break # Calculate chunk boundaries with overlap chunk_end = min(offset + args._scan_chunk_size, file_size) # Read chunk as bytes, decode to string chunk_bytes = mm[offset:chunk_end] try: chunk_text = chunk_bytes.decode("utf-8", errors="ignore") except Exception: chunk_text = chunk_bytes.decode("latin-1", errors="ignore") # Scan this chunk with all patterns for rx, pattern_str in patterns: try: for m in rx.finditer(chunk_text): # Calculate absolute position abs_start = offset + m.start() match_text = m.group(0) # Dedup matches at overlap boundaries match_key = (abs_start, match_text) if match_key in seen_matches: continue seen_matches.add(match_key) # Extract context ctx_start = max(0, m.start() - 60) ctx_end = min(len(chunk_text), m.end() + 60) context = chunk_text[ctx_start:ctx_end].replace("\n", " ") if is_false_positive(context, use_fp): continue results.append({ "severity": label, "source": url, "index": abs_start, "match": match_text, "pattern": pattern_str, "context": context.strip(), }) except re.error: continue # Skip problematic patterns # Advance with overlap to catch boundary matches if chunk_end >= file_size: break offset = chunk_end - OVERLAP_SIZE except (OSError, ValueError) as e: debug_log(f"[!] mmap failed for {file_path}: {e}, falling back to regular read\n") # Fallback for files that can't be mmap'd (e.g., special files) try: with open(file_path, "r", encoding="utf-8", errors="ignore") as f: body = f.read() for rx, pattern_str in patterns: for m in rx.finditer(body): start, end = m.start(), m.end() ctx_start = max(0, start - 60) ctx_end = min(len(body), end + 60) context = body[ctx_start:ctx_end].replace("\n", " ") if is_false_positive(context, use_fp): continue results.append({ "severity": label, "source": url, "index": start, "match": m.group(0), "pattern": pattern_str, "context": context.strip(), }) except OSError: pass return results def scan_files_python(file_map: Dict[str, str]) -> Dict[str, List[Dict[str, Any]]]: """ Python fallback scanner with support for gigabyte files. Uses chunked mmap for large files, parallel patterns for speed. """ results: Dict[str, List[Dict[str, Any]]] = {url: [] for url in file_map.values()} for path, url in file_map.items(): if STOP_EVENT.is_set(): break try: file_size = os.path.getsize(path) except OSError as e: debug_log(f"[!] Failed to stat {path} for {url}: {e}\n") continue # Use chunked scanning for large files if file_size > args._large_file_threshold: debug_log(f"[i] Large file ({file_size // (1024*1024)}MB): {url} - using chunked scan\n") for label, chunk in pattern_chunks: findings = scan_large_file_chunked(path, url, chunk, label, True) results[url].extend(findings) else: # Small files - load into memory for faster scanning try: with open(path, "r", encoding="utf-8", errors="ignore") as f: body = f.read() except OSError as e: debug_log(f"[!] Failed to read {path} for {url}: {e}\n") continue # Parallel pattern scanning for speed def scan_pattern_group(label_chunk): label, chunk = label_chunk group_results = [] for rx, pattern_str in chunk: try: for m in rx.finditer(body): start, end = m.start(), m.end() ctx_start = max(0, start - 60) ctx_end = min(len(body), end + 60) context = body[ctx_start:ctx_end].replace("\n", " ") if is_false_positive(context, True): continue group_results.append({ "severity": label, "source": url, "index": start, "match": m.group(0), "pattern": pattern_str, "context": context.strip(), }) except re.error: continue return group_results # Use thread pool for parallel pattern processing with ThreadPoolExecutor(max_workers=args._pattern_workers) as executor: futures = [executor.submit(scan_pattern_group, pc) for pc in pattern_chunks] for future in as_completed(futures): try: results[url].extend(future.result()) except Exception as e: debug_log(f"[!] Pattern scan error for {url}: {e}\n") return results # Config connect_timeout = args.timeout if args.timeout is not None else DEFAULT_CONNECT_TIMEOUT read_timeout = args.timeout if args.timeout is not None else DEFAULT_READ_TIMEOUT trust_env = True # Honor proxy environment variables user_agent = None if args.user_agent in (None, "") else args.user_agent # Use --max-file-size if specified, otherwise no limit (stream everything) max_body_bytes = args.max_file_size * 1024 * 1024 if args.max_file_size else DEFAULT_MAX_BODY_BYTES session_params = dict( retries=args.retries, backoff=1.0, # Default backoff factor trust_env=trust_env, pool_maxsize=pool_maxsize, ) total_urls = len(urls) + resumed_processed # Account for already processed URLs in resume mode progress_enabled = args.progress and total_urls > 0 and Progress is not None progress_cm = None progress = None progress_task = None if progress_enabled: progress_cm = Progress( SpinnerColumn(), BarColumn(bar_width=PROGRESS_BAR_WIDTH), MofNCompleteColumn(), TimeElapsedColumn(), TimeRemainingColumn(), TextColumn("{task.fields[url]}", justify="left"), transient=False, console=None, ) progress = progress_cm.__enter__() progress_task = progress.add_task("scan", total=total_urls, completed=resumed_processed, url="") # Determine output behavior: default overwrites once, then appends per write output_mode = "a" if args.output and not args.append: try: with open(args.output, "w", encoding="utf-8") as _: pass # truncate existing file for a fresh run except OSError as e: raise SystemExit(f"[!] Failed to open output file {args.output}: {e}") def advance_progress(url_msg: str): # Emit [progress] line for GUI (parsed by base-step.js) if args.machine_progress: try: sys.stderr.write(f"[progress] {processed + 1}/{total_urls}\n") sys.stderr.flush() except Exception: pass if not progress_enabled or progress is None or progress_task is None: return progress.update(progress_task, advance=1, url=url_msg) remaining = list(urls) processed = resumed_processed scan_completed = False try: batch_dir = temp_dir buffer_dir = temp_dir2 clear_dir(batch_dir) clear_dir(buffer_dir) batch_map: Dict[str, str] = {} buffer_map: Dict[str, str] = {} buffer_failed: List[str] = [] while remaining or batch_map or buffer_map: try: if STOP_EVENT.is_set(): break # Fill batch if empty if not batch_map and remaining: take = remaining[:max_parallel] remaining = remaining[max_parallel:] batch_map, failed_batch = fetch_batch(take, batch_dir) for url in failed_batch: advance_progress(f"{url} (failed)") processed += 1 # If this batch yielded nothing (all failed), try next chunk if not batch_map: continue if not batch_map and not remaining and not buffer_map: # nothing left anywhere break # Start rg on batch rg_results: Dict[str, List[Dict[str, Any]]] = {} rg_thread = ThreadPoolExecutor(max_workers=1) rg_future = rg_thread.submit(run_rg_on_dir, batch_dir, batch_map, pattern_files) # While rg runs, fill buffer up to nou buffer_map = {} if remaining: take_buf = remaining[:max_parallel] remaining = remaining[max_parallel:] buffer_map, buffer_failed = fetch_batch(take_buf, buffer_dir) for url in buffer_failed: advance_progress(f"{url} (failed)") processed += 1 rg_results = rg_future.result() rg_thread.shutdown(wait=True, cancel_futures=True) with RG_ENABLED_LOCK: rg_active = RG_ENABLED if not rg_active: rg_results = scan_files_python(batch_map) except KeyboardInterrupt: STOP_EVENT.set() try: rg_thread.shutdown(wait=False, cancel_futures=True) # type: ignore[name-defined] except Exception: pass break # Emit results per URL after rg finishes (dedupe per URL only) for url in batch_map.values(): findings = deduplicate_findings(rg_results.get(url, [])) if findings: if args.output: with open(args.output, output_mode, encoding="utf-8") as tf: tf.write(f"=== Results for {url} ===\n") for r in findings: tf.write(format_finding(r, color=False) + "\n") tf.write("\n") if not args.quiet: print(f"=== Results for {url} ===") for r in findings: print(format_finding(r, color=args.color)) print() advance_progress(url) processed += 1 clear_dir(batch_dir) # Swap batch and buffer for next cycle batch_dir, buffer_dir = buffer_dir, batch_dir batch_map = buffer_map buffer_map = {} buffer_failed = [] # Save state after each batch for resume capability # Combine remaining with buffer URLs that are still pending all_remaining = list(remaining) if buffer_map: all_remaining.extend(buffer_map.values()) save_scan_state(state_file, all_remaining, processed, args.output, args.list) # Mark scan as completed if we finished normally scan_completed = True except KeyboardInterrupt: STOP_EVENT.set() finally: if progress_enabled and progress is not None and progress_task is not None: try: progress.update(progress_task, completed=processed) except Exception: pass if progress_cm is not None: try: progress_cm.__exit__(None, None, None) except Exception: pass # Handle state file: clear on success, save on interruption if scan_completed: clear_scan_state(state_file) debug_log(f"[i] Scan completed. Processed {processed} URLs.\n") else: # Save final state for resume all_remaining = list(remaining) if batch_map: all_remaining.extend(batch_map.values()) if buffer_map: all_remaining.extend(buffer_map.values()) if all_remaining: save_scan_state(state_file, all_remaining, processed, args.output, args.list) debug_log( f"[i] Scan interrupted. Progress saved to {state_file}. " f"Use --resume to continue ({len(all_remaining)} URLs remaining).\n" ) if html_requested and html_output_path: html_script = os.path.join(os.path.dirname(os.path.realpath(__file__)), "html_secret_gen.py") cmd = [ sys.executable or "python3", html_script, "-i", args.output, "-o", html_output_path, "--max-blocks", str(args.max_blocks), ] try: subprocess.run(cmd, check=True) except subprocess.CalledProcessError as e: raise SystemExit(f"[!] Failed to generate HTML report: {e}") if __name__ == "__main__": main()