#!/usr/bin/env python3 """vercel-kv-scan — free Vercel KV / Postgres 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/vercel-kv-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": "vercel-kv-scan", "framework": "Vercel KV / Postgres", "detect": [".js", ".ts", ".tsx", ".jsx", ".mjs", "package.json", ".env", ".example"], "rules": [{"id": "vk-kv-import", "sev": "CRITICAL", "title": "@vercel/kv is sunset: move to @upstash/redis", "rx": "import\\s*\\{\\s*kv\\s*\\}\\s*from\\s*['\"]@vercel/kv['\"];?", "advice": "Use Redis.fromEnv() from @upstash/redis.", "fix": "import { Redis } from '@upstash/redis'; const kv = Redis.fromEnv();", "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "vk-kv-other", "sev": "HIGH", "title": "Other @vercel/kv import", "rx": "['\"]@vercel/kv['\"]", "advice": "Switch to @upstash/redis (new Redis({ url, token })).", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "vk-pg-import", "sev": "CRITICAL", "title": "@vercel/postgres is sunset: move to @neondatabase/serverless", "rx": "import\\s*\\{\\s*sql\\s*\\}\\s*from\\s*['\"]@vercel/postgres['\"];?", "advice": "Use neon(DATABASE_URL); note: results are rows[], not {rows}.", "fix": "import { neon } from '@neondatabase/serverless'; const sql = neon(process.env.DATABASE_URL);", "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "vk-pg-other", "sev": "HIGH", "title": "Other @vercel/postgres import (db/createPool/createClient)", "rx": "['\"]@vercel/postgres['\"]", "advice": "Use Pool/Client from @neondatabase/serverless.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "vk-rows", "sev": "MEDIUM", "title": "`.rows` access: neon() returns the rows array directly", "rx": "\\{\\s*rows\\s*\\}\\s*=\\s*await\\s+sql|\\.rows\\b", "advice": "Drop the `.rows` unwrap after switching to neon().", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "vk-env-kv", "sev": "MEDIUM", "title": "Legacy KV_REST_API_* env vars", "rx": "\\bKV_(?:REST_API_URL|REST_API_TOKEN|REST_API_READ_ONLY_TOKEN|URL)\\b", "advice": "Upstash Marketplace sets UPSTASH_REDIS_REST_URL/TOKEN; map them or re-link.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "vk-env-pg", "sev": "MEDIUM", "title": "Legacy POSTGRES_* env vars", "rx": "\\bPOSTGRES_(?:URL|PRISMA_URL|URL_NON_POOLING|USER|HOST|PASSWORD|DATABASE)\\b", "advice": "Neon provides DATABASE_URL (pooled) and DATABASE_URL_UNPOOLED.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "vk-pipeline", "sev": "MEDIUM", "title": "kv.pipeline()/multi(): Upstash pipelines are not atomic", "rx": "\\b(?:kv|redis)\\.(?:pipeline|multi)\\(", "advice": "Use multi() for transactions; exec() returns an array of results.", "fix": null, "flags": "", "unless": null, "ext": null, "fix_unless": null}, {"id": "vk-pool", "sev": "HIGH", "title": "Non-pooling connection string used from serverless code", "rx": "POSTGRES_URL_NON_POOLING|DATABASE_URL_UNPOOLED", "advice": "Use the pooled (-pooler) URL in functions; unpooled only for migrations.", "fix": null, "flags": "", "unless": null, "ext": [".js", ".ts", ".tsx", ".jsx", ".mjs"], "fix_unless": null}, {"id": "vk-dep", "sev": "HIGH", "title": "@vercel/kv or @vercel/postgres still in package.json", "rx": "\"@vercel/(?:kv|postgres)\"\\s*:", "advice": "Replace with @upstash/redis / @neondatabase/serverless.", "fix": null, "flags": "", "unless": null, "ext": ["package.json"], "fix_unless": null}], "samples": {"lib/cache.ts": "import { kv } from '@vercel/kv';\n\nexport async function hit(slug: string) {\n const p = kv.pipeline();\n p.incr(`views:${slug}`);\n p.expire(`views:${slug}`, 86400);\n return p.exec();\n}\n", "lib/db.ts": "import { sql } from '@vercel/postgres';\n\nexport async function getUser(id: string) {\n const { rows } = await sql`SELECT * FROM users WHERE id = ${id}`;\n return rows[0];\n}\nconst direct = process.env.POSTGRES_URL_NON_POOLING;\n", ".env.example": "KV_REST_API_URL=\nKV_REST_API_TOKEN=\nPOSTGRES_URL=\n", "package.json": "{ \"dependencies\": { \"@vercel/kv\": \"^1.0.1\", \"@vercel/postgres\": \"^0.9.0\" } }\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/fZucN72SO5oi0nO5qTgYU0t", "guide": "https://patchvault-81y.pages.dev/vercel-kv-postgres-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