`verification/fleet-classification/scan-schedulers.py`

136 lines
import os,re,json,plistlib,pathlib,subprocess,datetime,collections,urllib.parse,hashlib
HOME='/Users/macstudio3'; OUT=pathlib.Path('/tmp/tk11438-fleet-classification'); PROJECTS=HOME+'/Projects'
original=set(os.path.normpath(PROJECTS+'/'+x.strip()[2:]) for x in pathlib.Path(PROJECTS+'/tk11438-postgres-migration/verification/original-218-paths.txt').read_text().splitlines() if x.strip() not in ('','./'))
pm2=[]
try:
 for r in json.loads(pathlib.Path(HOME+'/.pm2/dump.pm2').read_text()):pm2.append({'name':r.get('name'),'cwd':r.get('pm_cwd'),'script':r.get('pm_exec_path')})
except Exception:pass
URL=re.compile(r'postgres(?:ql)?://[^\s\x22\x27`<>\\]+')
PG=re.compile(r'postgres(?:ql)?://|\b(?:psql|pg_dump|pg_restore|PGHOST|PGDATABASE|PGPORT|DATABASE_URL|KEN_DATABASE_URL|psycopg2?|asyncpg)\b|(?:require\([\x22\x27]pg[\x22\x27]|from [\x22\x27]pg[\x22\x27])',re.I)
EXT=('.sh','.bash','.zsh','.py','.mjs','.cjs','.js','.ts','.env')
cache={};read_errors={}
def norm(p,cwd=PROJECTS,variables=None):
 p=p.strip('"\'` ,;)').replace('${HOME}',HOME).replace('$HOME',HOME)
 if variables:
  for k,v in variables.items():p=p.replace('${'+k+'}',v).replace('$'+k+'/',v+'/')
 p=os.path.expanduser(p)
 if '$' in p or '://' in p or not p:return None
 return os.path.normpath(p if p.startswith('/') else os.path.join(cwd,p))
def load(p):
 if p in cache:return cache[p]
 try:
  f=pathlib.Path(p)
  if not f.is_file():raise ValueError('missing or not regular file')
  if f.stat().st_size>2_000_000:raise ValueError('larger than 2MB static scan limit')
  b=f.read_bytes()
  if b'\0' in b[:4096]:raise ValueError('binary executable not inspected')
  cache[p]=b.decode('utf8',errors='replace');return cache[p]
 except Exception as e:read_errors[p]=str(e) if isinstance(e,ValueError) else type(e).__name__;return None
def safe_target(value,key,source,line=None,context='literal'):
 try:
  u=urllib.parse.urlsplit(value);q=urllib.parse.parse_qs(u.query);host=q.get('host',[u.hostname or ''])[0];db=u.path.lstrip('/')
  transport='socket' if host.startswith('/') else 'local_tcp' if host in ('localhost','127.0.0.1','::1') else 'inherited_or_driver_default' if not host else 'dynamic_unresolved' if '$' in host or '{' in host else 'remote_or_container_tcp'
  return {'key':key,'source':source,'line':line,'host':host,'database':db,'port':str(u.port or 5432),'transport':transport,'context':context}
 except Exception:return {'key':key,'source':source,'line':line,'transport':'parse_unresolved','context':context}
def analyze(text,source,env=None):
 findings=[];envkeys=set((env or {}).keys());pg=bool(PG.search(text)) or any(PG.search(k) for k in envkeys)
 for n,line in enumerate(text.splitlines(),1):
  for m in URL.finditer(line):findings.append(safe_target(m.group(),None,source,n,'comment' if line.lstrip().startswith(('#','//')) else 'literal_or_fallback'))
  for m in re.finditer(r'\b((?:[A-Z][A-Z0-9_]*_)?(?:DATABASE_URL|DB_HOST|DB_NAME)|PGHOST|PGPORT|PGDATABASE)\b',line):envkeys.add(m.group())
  if pg:
   for m in re.finditer(r'\b(PGHOST|DB_HOST)\s*(?:=|:)\s*[\x22\x27]?([/\w.:-]+)',line):
    host=m.group(2);findings.append({'key':m.group(1),'source':source,'line':n,'host':host,'transport':'socket' if host.startswith('/') else 'local_tcp' if host in ('localhost','127.0.0.1','::1') else 'remote_or_unresolved','context':'static_assignment'})
 for k,v in (env or {}).items():
  if not isinstance(v,str):continue
  if URL.match(v):findings.append(safe_target(v,k,source,None,'scheduler_environment'))
  elif k in ('PGHOST','DB_HOST'):
   findings.append({'key':k,'source':source,'host':v,'transport':'socket' if v.startswith('/') else 'local_tcp' if v in ('localhost','127.0.0.1','::1') else 'remote_or_unresolved','context':'scheduler_environment'})
 if pg:
  for n,line in enumerate(text.splitlines(),1):
   if re.search(r'\b(?:psql|pg_dump|pg_restore)\b',line):
    h=re.search(r'(?:^|\s)(?:-h|--host(?:=|\s+))\s*[\x22\x27]?([/A-Za-z0-9_.:-]+)',line)
    d=re.search(r'(?:^|\s)(?:-d|--dbname(?:=|\s+))\s*[\x22\x27]?([A-Za-z0-9_.-]+)',line)
    if h:
     host=h.group(1);findings.append({'source':source,'line':n,'host':host,'database':d.group(1) if d else None,'transport':'socket' if host.startswith('/') else 'local_tcp' if host in ('localhost','127.0.0.1','::1') else 'remote_or_unresolved','context':'static_postgres_cli_host_candidate; remote shell context not executed'})
  for m in re.finditer(r'\bhost\s*:\s*[\x22\x27]([^\x22\x27]+)[\x22\x27]',text):
   host=m.group(1)
   if host in ('localhost','127.0.0.1','::1') or host.startswith('/tmp'):
    findings.append({'source':source,'line':text.count('\n',0,m.start())+1,'host':host,'transport':'socket' if host.startswith('/') else 'local_tcp','context':'host property in PostgreSQL-related source; config context not evaluated'})
 if pg and not findings:findings.append({'source':source,'transport':'inherited_or_driver_default','context':'Postgres indicator; no static literal target resolved'})
 return {'postgres_indicator':pg,'env_key_names':sorted(envkeys),'findings':findings}
def references(text,cwd,source=None):
 vars={'HOME':HOME}
 if source:vars.update({'DIR':os.path.dirname(source),'SCRIPT_DIR':os.path.dirname(source)})
 for k,v in re.findall(r'(?m)^\s*(?:export\s+)?([A-Z_]+)=[\x22\x27]?([^\n\x22\x27;]+)',text):
  p=norm(v,cwd,vars)
  if p and os.path.isdir(p):vars[k]=p
 found=set();unresolved=[]
 # Absolute/home/known-variable files and import/require literal relative modules.
 candidates=re.findall(r'(?:/Users/[^\s\x22\x27`<>;|)]+|/Library/[^\s\x22\x27`<>;|)]+|/opt/[^\s\x22\x27`<>;|)]+|~/[^\s\x22\x27`<>;|)]+|\$(?:\{?[A-Z_]+\}?)/[^\s\x22\x27`<>;|)]+)',text)
 candidates+=re.findall(r'(?:require\(|from\s+|import\s+)[\x22\x27](\.[^\x22\x27]+)',text)
 candidates+=re.findall(r'(?:\b(?:node|python3?|bash|zsh|source)\s+|(?:^|\s)\.\s+)[\x22\x27]?([^\s\x22\x27;|)]+)',text)
 for c in candidates:
  if not (c.endswith(EXT) or '/.env' in c or '/bin/' in c or '/scripts/' in c):continue
  p=norm(c,cwd,vars)
  if p and p!=source and '/node_modules/' not in p:found.add(p)
  elif not p:unresolved.append('dynamic_file_reference')
 if re.search(r'\b(?:npm|npx|pnpm|yarn)\b',text):unresolved.append('package_script_or_dynamic_command_not_expanded')
 if re.search(r'\bssh\b|\bdocker\b',text):unresolved.append('remote_or_container_command_not_executed')
 if re.search(r'\b(?:eval|exec)\s+\$|\b(?:bash|zsh)\s+-[lic]+\b',text):unresolved.append('dynamic_shell_or_login_environment')
 return found,sorted(set(unresolved))
rows=[];errors=[]
roots=[pathlib.Path(HOME+'/Library/LaunchAgents'),pathlib.Path('/Library/LaunchDaemons')]
for root in roots:
 for file in sorted(root.glob('*.plist')):
  try:p=plistlib.loads(file.read_bytes())
  except Exception as e:
   try:
    alt=subprocess.run(['plutil','-convert','json','-o','-',str(file)],capture_output=True,text=True)
    if alt.returncode:raise ValueError('plutil failed')
    p=json.loads(alt.stdout)
   except Exception:errors.append({'path':str(file),'error':type(e).__name__});continue
  if not isinstance(p,dict):errors.append({'path':str(file),'error':'plist root is not dictionary'});continue
  args=p.get('ProgramArguments',[]);program=p.get('Program') or (args[0] if args else None);cwd=p.get('WorkingDirectory') or HOME
  text='\n'.join(str(a) for a in args);a=analyze(text,str(file),p.get('EnvironmentVariables',{}))
  refs,unresolved=references(text,cwd)
  if program and pathlib.Path(program).suffix in EXT:refs.add(program)
  for x in args:
   if isinstance(x,str) and x.endswith(EXT) and not any(c in x for c in ['\n',';','|']):
    n=norm(x,cwd)
    if n:refs.add(n)
  reviewed=[];edges=[];queue=[(r,0) for r in sorted(refs)];seen=set()
  while queue and len(seen)<60:
   f,depth=queue.pop(0)
   if f in seen:continue
   seen.add(f);body=load(f)
   if body is None:edges.append({'source':f,'reason':read_errors[f]});continue
   item=analyze(body,f);item['source']=f;item['depth']=depth;item['sha256']=hashlib.sha256(body.encode()).hexdigest();item['original217']=f in original;reviewed.append(item)
   nxt,unknown=references(body,os.path.dirname(f),f);unresolved+=unknown
   if depth<2:queue.extend((x,depth+1) for x in sorted(nxt) if x not in seen)
   elif nxt:edges.extend({'source':x,'reason':'reference_depth_limit_2'} for x in sorted(nxt) if x not in seen)
  if queue:unresolved.append('per_job_file_limit_60')
  pg=a['postgres_indicator'] or any(x['postgres_indicator'] for x in reviewed)
  paths={r['source'] for r in reviewed}|refs
  related=[x['name'] for x in pm2 if x.get('script') in paths or (x.get('cwd') and any(z.startswith(x['cwd']+'/') for z in paths))]
  row={'label':p.get('Label'),'plist':str(file),'scheduler_domain':'user_launch_agents' if str(root).startswith(HOME) else 'system_launch_daemons','machine':'macstudio3 local scheduler; remote targets may be referenced','program':program,'working_directory':p.get('WorkingDirectory'),'start_interval':p.get('StartInterval'),'start_calendar_interval':p.get('StartCalendarInterval'),'run_at_load':p.get('RunAtLoad'),'disabled_in_file':p.get('Disabled',False),'loaded_state':'not_inspected; file presence does not prove loaded','env_key_names':sorted(p.get('EnvironmentVariables',{})),'plist_pg_findings':a['findings'],'postgres_candidate':pg,'referenced_sources':reviewed,'unreviewed_edges':edges,'dynamic_or_inherited_caveats':sorted(set(unresolved)),'original217_matches':sorted(paths&original),'saved_pm2_related_names':sorted(set(related))}
  if pg:row['inherited_environment_status']='Not resolved from loaded launchd domain, shell profile, .pg_service.conf, or all transitive imports; no jobs executed'
  rows.append(row)
cron=subprocess.run(['crontab','-l'],capture_output=True,text=True)
cr={'command':'crontab -l','exit_code':cron.returncode,'status':'no_user_crontab' if 'no crontab for' in cron.stderr else 'present' if cron.returncode==0 else 'unreviewed_error','lines':[]}
if cron.returncode==0:
 for n,line in enumerate(cron.stdout.splitlines(),1):
  if not line.strip() or line.lstrip().startswith('#'):continue
  fields=line.split(None,5);cr['lines'].append({'line':n,'schedule':fields[:5] if len(fields)==6 else [fields[0]],'postgres_analysis':analyze(line,'user_crontab:'+str(n)),'command_sha256':hashlib.sha256(line.encode()).hexdigest(),'command_not_executed':True})
counts=collections.Counter(f['transport'] for r in rows for f in r['plist_pg_findings']+[z for src in r['referenced_sources'] for z in src['findings']])
summary={'top_level_plists_discovered':sum(len(list(r.glob('*.plist'))) for r in roots),'top_level_plists_parsed':len(rows),'user_launch_agents':sum(r['scheduler_domain']=='user_launch_agents' for r in rows),'system_launch_daemons':sum(r['scheduler_domain']=='system_launch_daemons' for r in rows),'parse_errors':len(errors),'postgres_candidates':sum(r['postgres_candidate'] for r in rows),'unique_readable_referenced_sources':len(cache),'unique_unreadable_or_uninspected_sources':len(read_errors),'finding_occurrences_by_transport':dict(counts),'original217_unique_matches':len(set(z for r in rows for z in r['original217_matches'])),'nested_inactive_directory_plists_excluded':sum(1 for root in roots for f in root.rglob('*.plist') if f.parent!=root)}
runtime=OUT/'verification/runtime.json'
if runtime.exists():
 rt=json.loads(runtime.read_text())
 for row in rows:
  paths={r['source'] for r in row['referenced_sources']}
  row['current_pm2_related_names']=sorted(set(x['name'] for x in rt.get('services',[]) if x.get('script') in paths or (x.get('cwd') and any(z.startswith(x['cwd']+'/') for z in paths))))
summary['current_pm2_cross_reference_available']=runtime.exists()
result={'task':'TK-11438','owner':'codex-run-11438-verifier','timestamp':datetime.datetime.now(datetime.timezone.utc).isoformat(),'scope':'Static scheduler and bounded launcher source metadata; no execution or modifications','summary':summary,'coverage_limitations':['Only top-level plists in requested directories; /System/Library, other users, root crontab and remote schedulers not inspected','Static literal extraction includes comments/fallbacks; findings are candidates, not proof of effective runtime configuration','Up to two referenced-source hops and 60 files per scheduler; no eval, shell profile execution, dynamic command expansion or node module traversal','Saved and current PM2 metadata used only for path association, not proof of shared effective connection target','No loaded-state or effective inherited environment claims; no false-zero claim for jobs lacking static PG indicators'],'cron':cr,'errors':errors,'rows':rows}
(OUT/'schedulers.json').write_text(json.dumps(result,indent=2)+'\n')
print(json.dumps(summary,indent=2))