#!/usr/bin/env python3 """request-fetch-scan — free request → fetch migration & risk scanner by PatchVault (patchvault-81y.pages.dev). Single file, Python 3.8+ standard library only. Reads files locally; sends nothing anywhere. Usage: python3 <(curl -sL https://patchvault-81y.pages.dev/starters/request-fetch-scan.py) [dir] [--demo] [--json] [--no-color] If the directory has no matching project files (or --demo is given) it runs a clearly labelled showcase on a built-in SAMPLE repository so you can see the report format. Exit code is always 0 (it is a report, not a gate).""" import json,os,re,sys CFG=json.loads(r'''{"name": "request-fetch-scan", "framework": "request → fetch", "detect": [".js", ".cjs", ".mjs", ".ts", ".tsx", ".jsx", "package.json"], "rules": [{"id": "rf-require", "sev": "CRITICAL", "title": "Deprecated `request` package (unmaintained since 2020, CVE-2023-28155)", "rx": "require\\(\\s*['\"]request['\"]\\s*\\)", "advice": "Point the import at the drop-in fetch shim (lib/request-fetch.cjs).", "fix": "require('./lib/request-fetch.cjs')", "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-import", "sev": "CRITICAL", "title": "ESM import of deprecated `request`", "rx": "from\\s+['\"]request['\"]", "advice": "Import the fetch shim instead.", "fix": "from './lib/request-fetch.mjs'", "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-promise", "sev": "HIGH", "title": "request-promise / request-promise-native (deprecated with request)", "rx": "['\"]request-promise(?:-native)?['\"]", "advice": "Use the shim requestPromise export or native fetch.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-pipe", "sev": "HIGH", "title": "Stream .pipe() on a request call: fetch returns a web ReadableStream", "rx": "request(?:\\.\\w+)?\\([^)]*\\)\\s*\\.pipe\\(", "advice": "Use Readable.fromWeb((await fetch(url)).body).pipe(dest).", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-formdata", "sev": "MEDIUM", "title": "multipart formData option: fetch needs a FormData body (no manual Content-Type)", "rx": "\\bformData\\s*:", "advice": "Build `new FormData()`; let fetch set the multipart boundary.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-timeout", "sev": "MEDIUM", "title": "`timeout:` option has no fetch equivalent", "rx": "\\btimeout\\s*:\\s*\\d+", "advice": "Use `signal: AbortSignal.timeout(ms)`.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-retry", "sev": "MEDIUM", "title": "Custom retry options (requestretry style) are ignored by fetch", "rx": "\\b(?:maxAttempts|retryDelay|retryStrategy)\\b|requestretry", "advice": "Wrap fetch in a small retry loop with backoff.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-oauth", "sev": "HIGH", "title": "`oauth:` signing option (OAuth 1.0a) has no fetch equivalent", "rx": "\\boauth\\s*:\\s*\\{", "advice": "Sign the Authorization header yourself (e.g. oauth-1.0a) before fetch.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-auth", "sev": "MEDIUM", "title": "`auth:` option: fetch needs an explicit Authorization header", "rx": "\\bauth\\s*:\\s*\\{\\s*(?:user|username|bearer|pass)", "advice": "Set headers.Authorization (Basic base64 / Bearer).", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-json", "sev": "MEDIUM", "title": "`json: true`: fetch needs JSON.stringify + await res.json()", "rx": "\\bjson\\s*:\\s*true\\b", "advice": "Send `body: JSON.stringify(x)` with content-type and parse with `await res.json()`.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-tls", "sev": "HIGH", "title": "TLS verification disabled", "rx": "strictSSL\\s*:\\s*false|rejectUnauthorized\\s*:\\s*false", "advice": "Remove it; fix the certificate chain instead.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "rf-dep", "sev": "HIGH", "title": "`request` still listed in package.json", "rx": "\"request(?:-promise(?:-native)?)?\"\\s*:\\s*\"", "advice": "Remove the dependency after the shim is in place.", "fix": null, "flags": "", "unless": null, "ext": ["package.json"], "fix_unless": null}], "samples": {"src/api/github.js": "const request = require('request');\nconst rp = require('request-promise-native');\n\nfunction getRepo(owner, name, cb) {\n request({\n url: `https://api.github.com/repos/${owner}/${name}`,\n json: true,\n timeout: 5000,\n auth: { bearer: process.env.GH_TOKEN },\n headers: { 'User-Agent': 'acme-bot' },\n }, cb);\n}\nmodule.exports = { getRepo, rp };\n", "src/upload.js": "const request = require('request');\nconst fs = require('fs');\n\nexports.upload = (file) => request.post({\n url: 'https://files.acme.dev/upload',\n formData: { file: fs.createReadStream(file) },\n strictSSL: false,\n});\nexports.mirror = (url, dest) => request.get(url).pipe(fs.createWriteStream(dest));\n", "package.json": "{\n \"name\": \"acme-api\",\n \"dependencies\": {\n \"request\": \"^2.88.2\",\n \"request-promise-native\": \"^1.0.9\"\n }\n}\n"}, "hours": {"CRITICAL": 1.5, "HIGH": 1.0, "MEDIUM": 0.5}, "hours_note": "Estimate: 1.5 h per CRITICAL, 1.0 h per HIGH, 0.5 h per MEDIUM issue (find, rewrite, re-test by hand).", "buy": "https://buy.stripe.com/00w9AV0KGaIC5I86uXgYU0o", "guide": "https://patchvault-81y.pages.dev/request-to-fetch-migration/?src=cli_terminal", "price": 9}''') SKIP={'node_modules','.git','dist','build','.next','coverage','.vercel','.venv','venv','__pycache__','.turbo','.cache'} SEVS=('CRITICAL','HIGH','MEDIUM') def _rules(): out=[] for r in CFG['rules']: out.append(dict(r,_rx=re.compile(r['rx'],re.I if 'i' in r['flags'] else 0),_un=re.compile(r['unless'],re.I) if r['unless'] else None,_fu=re.compile(r['fix_unless']) if r.get('fix_unless') else None)) return out def matches_detect(rel): b=os.path.basename(rel) return any(b==d or (d.startswith('.') and (b.endswith(d) or b.startswith(d+'.') or b==d)) for d in CFG['detect']) def rule_applies(r,rel): if not r['ext']: return True b=os.path.basename(rel); return any(b==e or b.endswith(e) or (e.startswith('.') and b.startswith(e)) for e in r['ext']) def fix_line(rel,line,R): for r in R: if r['fix'] is None or not rule_applies(r,rel) or (r['_fu'] and r['_fu'].search(line)): continue line=r['_rx'].sub(r['fix'],line) return line def scan_text(rel,text,R): F=[]; lines=text.splitlines() for r in R: if not rule_applies(r,rel) or (r['_un'] and r['_un'].search(text)): continue for n,l in enumerate(lines,1): if r['_rx'].search(l): F.append(dict(sev=r['sev'],id=r['id'],file=rel,line=n,title=r['title'],advice=r['advice'],code=l.rstrip(), fixed=fix_line(rel,l.rstrip(),R) if r['fix'] is not None and not (r['_fu'] and r['_fu'].search(l)) else None,ctx=lines)) return F def collect(root): files=[] for dp,dn,fn in os.walk(root): dn[:]=sorted(x for x in dn if x not in SKIP) for f in sorted(fn): p=os.path.join(dp,f); rel=os.path.relpath(p,root) if f in ('package-lock.json','yarn.lock','pnpm-lock.yaml') or not matches_detect(rel): continue try: if os.path.getsize(p)>2_000_000: continue files.append((rel,open(p,encoding='utf-8',errors='ignore').read())) except OSError: pass return files class C: def __init__(s,on): s.on=on def __call__(s,code,t): return f'\033[{code}m{t}\033[0m' if s.on else t def vis(t): return len(re.sub(r'\033\[[0-9;]*m','',t)) def main(a): try: sys.stdout.reconfigure(errors='replace') except Exception: pass root=next((x for x in a if not x.startswith('-')),'.') uni=(sys.stdout.encoding or '').lower().startswith('utf') c=C(sys.stdout.isatty() and not os.environ.get('NO_COLOR') and '--no-color' not in a and os.environ.get('TERM')!='dumb') R=_rules(); files=[] if '--demo' in a else collect(root); demo=not files if demo: files=list(CFG['samples'].items()) F=[f for rel,t in files for f in scan_text(rel,t,R)] F.sort(key=lambda f:(SEVS.index(f['sev']),f['file'],f['line'])) cnt={s:sum(f['sev']==s for f in F) for s in SEVS} hrs=sum(CFG['hours'][f['sev']] for f in F) buy=CFG['buy']+('&' if '?' in CFG['buy'] else '?')+'src=cli_terminal' if '--json' in a: print(json.dumps(dict(tool=CFG['name'],sample_mode=demo,files_scanned=len(files),counts=cnt,est_manual_hours=hrs,hours_note=CFG['hours_note'], findings=[{k:f[k] for k in ('sev','id','file','line','title','advice','code','fixed')} for f in F],guide=CFG['guide'],fix_bundle=buy),indent=1)); return 0 H,V,TL,TR,BL,BR,LT,RT='─│┌┐└┘├┤' if uni else '-|++++++' badge={'CRITICAL':c('1;97;41',' CRITICAL '),'HIGH':c('1;30;43',' HIGH '),'MEDIUM':c('1;30;46',' MEDIUM ')} print() print(c('1;33',f"⚡ PatchVault {CFG['name']}" if uni else f"PatchVault {CFG['name']}")+c('2',f" · {CFG['framework']}")) if demo: print(c('1;35',f"⚡ No local {CFG['framework']} files detected in current directory — running Live Interactive Showcase on sample repository..." if uni else f"No local {CFG['framework']} files detected in current directory - running Live Interactive Showcase on sample repository...")) print(c('2',' (SAMPLE DATA: the files below are a built-in demo repo, not your code. Run inside your project to scan it.)')) W=64 rows=[('Files scanned',str(len(files))+(' (sample repo)' if demo else '')), ('[CRITICAL] issues',c('1;91',str(cnt['CRITICAL']))),('[HIGH] issues',c('1;93',str(cnt['HIGH']))),('[MEDIUM] issues',c('1;96',str(cnt['MEDIUM']))), ('Est. manual fix time',f'~{hrs:g} h (estimate)'),('PatchVault auto-fix',c('1;92','~12 seconds')+f" ({sum(f['fixed'] is not None for f in F)} auto-fixable)")] print(TL+H*W+TR) t=' SCAN SUMMARY'; print(V+c('1',t)+' '*(W-len(t))+V); print(LT+H*W+RT) for k,v in rows: line=f' {k:<24}{v}'; print(V+line+' '*max(0,W-vis(line))+V) print(BL+H*W+BR) print(c('2',' '+CFG['hours_note'])) if not F: print('\n '+c('1;92','Nothing found. You look ready.')) else: print() for f in F[:60]: print(f" {badge[f['sev']]} {c('1',f['file'])}:{f['line']} {f['title']}") if len(F)>60: print(f' ... and {len(F)-60} more (use --json)') f=F[0]; L=f['ctx']; i=f['line']-1 new=f['fixed'] if f['fixed'] is not None and f['fixed']!=f['code'] else re.match(r'\s*',f['code']).group(0)+('# ' if f['file'].endswith(('.toml','.sh','.yml','.yaml','.py','.env','.example')) else '-- ' if f['file'].endswith('.sql') else '// ')+'PatchVault: '+f['advice'] print('\n'+c('1',f" Before/after preview — first issue ({f['id']})"+(' [sample]' if demo else ''))) print(c('1',f" --- a/{f['file']}")); print(c('1',f" +++ b/{f['file']}")) print(' '+c('2',f"@@ line {f['line']} @@ "+(L[i-1].strip()[:70] if i>0 else ''))) print(' '+c('91','- '+f['code'][:110])); print(' '+c('92','+ '+new[:110])) print(' '+c('2',' '+(L[i+1][:110] if i+1