TagTrawler.py

#!/usr/bin/env python3
"""
TagTrawler - scan Docker images (local or remote) for likely secrets.

Key features:
- --remote one or more image repos/uris
- --all-tags enumerates tags (Docker Hub, ECR, GCR) and scans each tag
- Registry auto-detection, with --registry override
- Parallel tag scanning via --threads
- Optional regex config (JSON/YAML) for filename and content patterns
- Outputs structured JSON "table-like" rows (one per image:tag)

Auth philosophy:
- Image pulling uses whatever Docker is already logged into (docker login, credential helpers, hub-tool, gcloud helper, etc.)
- Tag enumeration:
  - Docker Hub: unauthenticated where possible; if 401/404 and creds exist in Docker config, uses JWT login.
  - GCR: uses Docker credential helper token (docker-credential-gcloud) if configured.
  - ECR: uses boto3 (AWS creds/role/SSO/etc.); region parsed from ECR hostname.
"""

from __future__ import annotations

import argparse
import base64
import concurrent.futures as cf
import dataclasses
import datetime as dt
import json
import os
import re
import subprocess
import tarfile
import tempfile
import threading
from typing import Dict, List, Optional, Tuple

import boto3
import docker
import requests

try:
    import yaml  # PyYAML
except Exception:
    yaml = None

# =========================
# Defaults
# =========================

# =========================
# Linux container noise reduction (mode: linux-container)
# =========================

LINUX_CONTAINER_EXCLUDE_PATH_REGEX = [
    r"^/usr/share/doc(/|$)",
    r"^/usr/share/man(/|$)",
    r"^/usr/share/info(/|$)",
    r"^/usr/share/locale(/|$)",
    r"^/var/lib/dpkg(/|$)",
    r"^/var/cache/apt(/|$)",
    r"^/var/cache/dnf(/|$)",
    r"^/var/cache/yum(/|$)",
    r"^/var/log(/|$)",
    r"^/proc(/|$)",
    r"^/sys(/|$)",
    r"^/dev(/|$)",
    r"^/run(/|$)",  # note: secrets often in /run/secrets; handled by deep roots below
    r"^/usr/src(/|$)",
]

# Skip content scanning for these extensions by default (still allow filename hits)
LINUX_CONTAINER_EXCLUDE_EXTENSIONS = {
    # images/fonts/media/static
    ".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico", ".webp", ".bmp",
    ".ttf", ".otf", ".woff", ".woff2", ".eot",
    ".mp3", ".mp4", ".mov", ".avi", ".mkv", ".wav",
    ".css", ".map",
    # compiled/binary artifacts
    ".so", ".a", ".o", ".obj", ".dll", ".exe",
    ".pyc", ".pyo", ".class",
    # package payloads
    ".deb", ".rpm", ".apk", ".whl",
}

# Only content-scan these extensions by default (plus small, text-like no-ext files)
LINUX_CONTAINER_CONTENT_EXT_ALLOWLIST = {
    ".env", ".conf", ".ini", ".cfg", ".cnf",
    ".yaml", ".yml", ".json", ".toml", ".properties",
    ".tf", ".tfvars",
    ".sh", ".bash", ".zsh", ".ksh",
    ".py", ".js", ".ts", ".go", ".rb", ".php", ".java", ".cs", ".c", ".cpp", ".h", ".hpp",
    ".sql",
    ".txt", ".md", ".log",
    ".pem", ".key", ".ppk", ".p12", ".pfx",
}

# Deep-scan roots where custom app configs and secrets usually live
LINUX_CONTAINER_DEEP_ROOTS = (
    "/app", "/opt", "/srv", "/workspace", "/var/www",
    "/home", "/root",
    "/usr/local",  # common for custom installs
    "/run/secrets", "/var/run/secrets",
)

# System trees: avoid deep content scanning unless file is config-like
LINUX_CONTAINER_SYSTEM_ROOTS = ("/usr/bin", "/usr/sbin", "/usr/lib", "/lib", "/lib64")

BENIGN_CONTEXT_RE = re.compile(r"(?i)\b(example|sample|placeholder|changeme|dummy|lorem|test)\b")
DEFAULT_FILENAME_PATTERNS = [
    r"(^|/)\.env(\..*)?$",
    r"(^|/)(id_rsa|id_dsa|id_ecdsa|id_ed25519)$",
    r"(^|/)(authorized_keys)$",
    r"(^|/)(kubeconfig|\.kube/config)$",
    r"(^|/)\.aws/credentials$",
    r"(^|/)\.npmrc$",
    r"(^|/)\.pypirc$",
    r"(^|/)\.dockercfg$",
    r"(^|/)(config\.json|settings\.json)$",
    r"(^|/)(credential|credentials)([._-]|$)",
    r"(^|/)(secret|secrets)([._-]|$)",
    r"(^|/)(api[_-]?key|apikey)([._-]|$)",
    r"(^|/)(token|tokens)([._-]|$)",
    r"(^|/)(pass(word)?|passwd)([._-]|$)",
    r"(^|/).*private([._-]|)key.*",
    r"\.kdbx$",
    r"\.rdp$",
    r"\.(vhd|vhdx|vmdk|qcow2|img|iso|ova|ovf)$",
    r"\.(bak|old|orig|backup)$",
    r"\.(sql|sqlite|db|dmp)$",
    r"\.(zip|7z|rar|tar|tgz|gz)$",
    r"\.(pem|key|ppk|pfx|p12)$",
]

DEFAULT_CONTENT_PATTERNS = [
    r"AKIA[0-9A-Z]{16}",
    r"ASIA[0-9A-Z]{16}",
    r"(?i)aws(.{0,20})?(secret|access)[_-]?key['\"\s:=]+[A-Za-z0-9/+=]{16,}",
    r"AIza[0-9A-Za-z\-_]{35}",
    r"ya29\.[0-9A-Za-z\-_]+",
    r"(?i)ghp_[A-Za-z0-9]{20,}",
    r"(?i)github_pat_[A-Za-z0-9_]{20,}",
    r"(?i)glpat-[A-Za-z0-9\-_]{20,}",
    r"(?i)xox[baprs]-[A-Za-z0-9-]{10,48}",
    r"(?i)sk_(live|test)_[A-Za-z0-9]{16,}",
    r"(?i)rk_(live|test)_[A-Za-z0-9]{16,}",
    r"(?i)SG\.[A-Za-z0-9_\-]{16,}\.[A-Za-z0-9_\-]{16,}",
    r"(?i)eyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}",
    r"-----BEGIN (RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----",
    r"(?i)authorization:\s*basic\s+[A-Za-z0-9\+/=]{12,}",
    r"(?i)bearer\s+[A-Za-z0-9\-\._~\+\/]+=*",
    r"(?i)\b(api[_-]?key|apikey)\b\s*[:=]\s*['\"]?[A-Za-z0-9_\-\.=\/\+]{20,}['\"]?",
    r"(?i)\b(token|access[_-]?token|refresh[_-]?token)\b\s*[:=]\s*['\"]?[A-Za-z0-9_\-\.=\/\+]{20,}['\"]?",
    r"(?i)\b(client[_-]?secret|secret)\b\s*[:=]\s*['\"]?[A-Za-z0-9_\-\.=\/\+]{16,}['\"]?",
    r"(?i)\b(pass(word)?|passwd)\b\s*[:=]\s*['\"]?[^\s'\";]{10,}['\"]?",
]

DEFAULT_SCAN_PATHS = ["/"]  # export root by default; can be overridden

DOCKERHUB_INDEX_KEYS = [
    "https://index.docker.io/v1/",
    "index.docker.io",
    "registry-1.docker.io",
    "https://registry-1.docker.io",
]

# =========================
# Utilities
# =========================

INTERESTING_ENV_NAME_RE = re.compile(
    r"(?i)(KEY|TOKEN|SECRET|PASS|PWD|CRED|CONN|SAS|SIGNATURE|PRIVATE|CERT|KUBE|AZURE|AWS|GCP|GOOGLE|GITHUB|GITLAB|SLACK|STRIPE|SENDGRID|TWILIO|URL|URI|ENDPOINT|HOST|SERVER)"
)

def filter_interesting_env(env_list: List[str]) -> List[str]:
    out = []
    for kv in env_list:
        k = kv.split("=", 1)[0] if "=" in kv else kv
        if INTERESTING_ENV_NAME_RE.search(k):
            out.append(kv)
    return out

def utc_now_compact() -> str:
    return dt.datetime.now(dt.timezone.utc).strftime("%Y%m%d_%H%M%S")

def eprint(msg: str) -> None:
    print(msg, flush=True)

def guess_docker_config_path() -> str:
    docker_config = os.getenv("DOCKER_CONFIG")
    if docker_config:
        return os.path.join(docker_config, "config.json")
    return os.path.expanduser("~/.docker/config.json")

def load_docker_config() -> Dict:
    path = guess_docker_config_path()
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}

def call_credential_helper(helper: str, server: str) -> Optional[Tuple[str, str]]:
    """
    Calls docker-credential-<helper> get, returns (username, secret) if available.
    """
    exe = f"docker-credential-{helper}"
    try:
        proc = subprocess.run(
            [exe, "get"],
            input=server.encode("utf-8"),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )
        if proc.returncode != 0:
            return None
        data = json.loads(proc.stdout.decode("utf-8", errors="ignore"))
        user = data.get("Username")
        secret = data.get("Secret")
        if user is None or secret is None:
            return None
        return user, secret
    except Exception:
        return None

def get_registry_creds_from_docker_config(registry: str) -> Optional[Tuple[str, str]]:
    """
    Try to resolve credentials for a registry from ~/.docker/config.json.
    Works with:
      - inline auth (base64 username:password)
      - credsStore
      - credHelpers
    """
    cfg = load_docker_config()
    auths = cfg.get("auths", {}) or {}
    # registry keys can be stored with or without scheme. Try common variants.
    candidates = [registry, f"https://{registry}", f"http://{registry}"]
    # include known dockerhub keys if registry is docker hub-ish
    if registry in {"index.docker.io", "registry-1.docker.io", "docker.io"}:
        candidates.extend(DOCKERHUB_INDEX_KEYS)

    # 1) Inline auths
    for key in candidates:
        entry = auths.get(key)
        if not isinstance(entry, dict):
            continue
        auth_b64 = entry.get("auth")
        if auth_b64:
            try:
                raw = base64.b64decode(auth_b64).decode("utf-8", errors="ignore")
                if ":" in raw:
                    user, pw = raw.split(":", 1)
                    if user and pw:
                        return user, pw
            except Exception:
                pass

    # 2) Cred helpers
    cred_helpers = cfg.get("credHelpers", {}) or {}
    creds_store = cfg.get("credsStore")

    # Prefer per-registry helper
    helper = None
    for key in candidates:
        if key in cred_helpers:
            helper = cred_helpers.get(key)
            break
    if helper is None and creds_store:
        helper = creds_store

    if helper:
        # Docker helpers often use registry host without scheme as "server"
        for server in candidates + [registry]:
            creds = call_credential_helper(helper, server)
            if creds:
                return creds

    return None

@dataclasses.dataclass(frozen=True)
class RegexConfig:
    filename_patterns: List[re.Pattern]
    content_patterns: List[re.Pattern]

def compile_patterns(patterns: List[str], ignore_case: bool = False) -> List[re.Pattern]:
    flags = re.IGNORECASE if ignore_case else 0
    out = []
    for p in patterns:
        out.append(re.compile(p, flags))
    return out

def load_regex_config(path: Optional[str]) -> RegexConfig:
    if not path:
        return RegexConfig(
            filename_patterns=compile_patterns(DEFAULT_FILENAME_PATTERNS, ignore_case=True),
            content_patterns=compile_patterns(DEFAULT_CONTENT_PATTERNS, ignore_case=False),
        )
    try:
        with open(path, "r", encoding="utf-8") as f:
            if path.endswith((".yml", ".yaml")):
                if yaml is None:
                    raise RuntimeError("PyYAML not installed but YAML config provided.")
                data = yaml.safe_load(f)
            else:
                data = json.load(f)
        fp = data.get("filename_patterns", DEFAULT_FILENAME_PATTERNS)
        cp = data.get("content_patterns", DEFAULT_CONTENT_PATTERNS)
        return RegexConfig(
            filename_patterns=compile_patterns(fp, ignore_case=True),
            content_patterns=compile_patterns(cp, ignore_case=False),
        )
    except Exception as e:
        eprint(f"[!] Failed to load regex config '{path}': {e}. Using defaults.")
        return RegexConfig(
            filename_patterns=compile_patterns(DEFAULT_FILENAME_PATTERNS, ignore_case=True),
            content_patterns=compile_patterns(DEFAULT_CONTENT_PATTERNS, ignore_case=False),
        )

def detect_registry(image_ref: str) -> str:
    """
    Returns one of: dockerhub, ecr, gcr, acr, other
    """
    # Split registry host if present
    # image_ref can be: namespace/repo or host/namespace/repo
    parts = image_ref.split("/")
    if len(parts) >= 2 and ("." in parts[0] or ":" in parts[0]):
        host = parts[0]
        if host.endswith("amazonaws.com"):
            return "ecr"
        if host.endswith("azurecr.io"):
            return "acr"
        if host.endswith("gcr.io") or host.endswith("pkg.dev") or host.endswith("googleusercontent.com") or host in {"gcr.io", "us.gcr.io", "eu.gcr.io", "asia.gcr.io"}:
            return "gcr"
        return "other"
    # No explicit host: assume Docker Hub
    return "dockerhub"

def split_image_and_tag(image: str) -> Tuple[str, Optional[str]]:
    """
    Split <ref>[:tag] while not breaking host:port.
    """
    # If tag is present, it's after last ':' that occurs AFTER last '/'
    last_colon = image.rfind(":")
    last_slash = image.rfind("/")
    if last_colon > last_slash:
        return image[:last_colon], image[last_colon+1:]
    return image, None

# =========================
# Tag listing implementations
# =========================

def dockerhub_list_tags(repo: str, session: requests.Session, verbose: bool, auth: Optional[Tuple[str, str]] = None) -> List[str]:
    """
    Docker Hub tag listing (public + private) using the Docker Registry (distribution) API:

      1) Get a Bearer token from auth.docker.io for scope repository:<name>:pull
      2) Call https://registry-1.docker.io/v2/<name>/tags/list

    This avoids https://hub.docker.com/v2/users/login/ which can return 500s.

    repo: "namespace/name" or "name" (official images assumed under library/)
    auth: (username, password/token) if available (from Docker config/credential helper)
    """
    if "/" not in repo:
        name = f"library/{repo}"
    else:
        name = repo

    # Step 1: token
    token_url = (
        "https://auth.docker.io/token"
        f"?service=registry.docker.io&scope=repository:{name}:pull"
    )
    try:
        if auth:
            if verbose:
                eprint(f"[i] Docker Hub: requesting token for {name} with docker-stored creds")
            tok_resp = session.get(token_url, auth=auth, timeout=30)
        else:
            if verbose:
                eprint(f"[i] Docker Hub: requesting anonymous token for {name}")
            tok_resp = session.get(token_url, timeout=30)

        if tok_resp.status_code != 200:
            if verbose:
                eprint(f"[!] Docker Hub token request failed ({tok_resp.status_code}) for {name}: {tok_resp.text[:200]}")
            return []
        tok_json = tok_resp.json()
        token = tok_json.get("token") or tok_json.get("access_token")
        if not token:
            if verbose:
                eprint(f"[!] Docker Hub token response missing token for {name}")
            return []
    except Exception as e:
        if verbose:
            eprint(f"[!] Docker Hub token request error for {name}: {e}")
        return []

    # Step 2: tags/list with best-effort pagination ('n' + 'last')
    tags: List[str] = []
    base = f"https://registry-1.docker.io/v2/{name}/tags/list"
    n = 1000
    last = None

    while True:
        params = {"n": str(n)}
        if last:
            params["last"] = last
        try:
            resp = session.get(base, headers={"Authorization": f"Bearer {token}"}, params=params, timeout=30)
            if resp.status_code != 200:
                if verbose:
                    eprint(f"[!] Docker Hub tags/list failed ({resp.status_code}) for {name}: {resp.text[:200]}")
                break
            data = resp.json()
            batch = data.get("tags") or []
            if not batch:
                break
            tags.extend(batch)
            if len(batch) < n:
                break
            last = batch[-1]
        except Exception as e:
            if verbose:
                eprint(f"[!] Docker Hub tags/list error for {name}: {e}")
            break

    return sorted(set(tags))

def ecr_list_tags(image_ref: str, verbose: bool) -> List[str]:
    """
    image_ref includes full ECR host: <acct>.dkr.ecr.<region>.amazonaws.com/<repo>
    """
    image_no_tag, _ = split_image_and_tag(image_ref)
    host, repo = image_no_tag.split("/", 1)
    # host looks like: 123456789012.dkr.ecr.us-east-1.amazonaws.com
    parts = host.split(".")
    registry_id = parts[0]
    # region is after 'ecr'
    region = None
    for i, p in enumerate(parts):
        if p == "ecr" and i + 1 < len(parts):
            region = parts[i + 1]
            break
    if region is None:
        region = os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "us-east-1"
        if verbose:
            eprint(f"[i] ECR: could not parse region from host; defaulting to {region}")

    ecr = boto3.client("ecr", region_name=region)
    tags: List[str] = []
    try:
        paginator = ecr.get_paginator("describe_images")
        for page in paginator.paginate(repositoryName=repo, registryId=registry_id):
            for detail in page.get("imageDetails", []) or []:
                for t in detail.get("imageTags", []) or []:
                    tags.append(t)
    except Exception as e:
        if verbose:
            eprint(f"[!] ECR tag listing failed for {image_ref}: {e}")
    return sorted(set(tags))

def gcr_list_tags(image_ref: str, verbose: bool) -> List[str]:
    """
    Uses Docker Registry HTTP API for GCR-like registries.
    Attempts to use docker credential helper (e.g., docker-credential-gcloud) if configured.

    image_ref example: gcr.io/project/image or us.gcr.io/project/image or <region>-docker.pkg.dev/project/repo/image
    """
    image_no_tag, _ = split_image_and_tag(image_ref)
    host, path = image_no_tag.split("/", 1)
    # Registry v2 endpoint
    url = f"https://{host}/v2/{path}/tags/list"
    headers = {}

    # Try to obtain creds/token from docker config helper
    creds = get_registry_creds_from_docker_config(host)
    if creds:
        user, secret = creds
        # Many helpers return username "_token" and secret as access token
        # We'll try Bearer token first if username suggests token.
        if verbose:
            eprint(f"[i] GCR: found docker creds for {host} (username={user})")
        if user in {"_token", "oauth2accesstoken"}:
            headers["Authorization"] = f"Bearer {secret}"
        else:
            # Basic auth (rare for GCR); send Authorization header
            b64 = base64.b64encode(f"{user}:{secret}".encode("utf-8")).decode("utf-8")
            headers["Authorization"] = f"Basic {b64}"

    try:
        resp = requests.get(url, headers=headers, timeout=30)
        if resp.status_code != 200:
            if verbose:
                eprint(f"[!] GCR tags/list failed ({resp.status_code}) for {image_no_tag}")
            return []
        data = resp.json()
        tags = data.get("tags", []) or []
        return sorted(set(tags))
    except Exception as e:
        if verbose:
            eprint(f"[!] GCR tag listing error for {image_no_tag}: {e}")
        return []


def _registry_bearer_token(host: str, repo_path: str, scope_action: str, session: requests.Session,
                           verbose: bool) -> Optional[str]:
    """
    Generic Docker Registry v2 Bearer token acquisition:
      - GET https://<host>/v2/ to obtain WWW-Authenticate challenge
      - Request token from the challenge realm with service + scope
    Uses credentials from Docker config/credential helpers when available.
    """
    try:
        # Trigger auth challenge
        r = session.get(f"https://{host}/v2/", timeout=30)
        if r.status_code in (200, 401):
            www = r.headers.get("Www-Authenticate") or r.headers.get("WWW-Authenticate")
        else:
            www = r.headers.get("Www-Authenticate") or r.headers.get("WWW-Authenticate")
        if not www or "Bearer" not in www:
            if verbose:
                eprint(f"[!] Registry auth challenge missing Bearer header for {host} (status={r.status_code})")
            return None

        # Parse: Bearer realm="...",service="...",scope="..."
        # We'll extract key="value" pairs.
        realm = None
        service = None
        params = {}
        parts = www.split(" ", 1)
        if len(parts) == 2:
            kvs = parts[1].split(",")
            for kv in kvs:
                kv = kv.strip()
                if "=" in kv:
                    k, v = kv.split("=", 1)
                    v = v.strip().strip('"')
                    params[k.strip()] = v
        realm = params.get("realm")
        service = params.get("service")
        if not realm:
            if verbose:
                eprint(f"[!] Could not parse auth realm from WWW-Authenticate for {host}: {www}")
            return None

        scope = f"repository:{repo_path}:{scope_action}"
        token_params = {"service": service, "scope": scope} if service else {"scope": scope}

        creds = get_registry_creds_from_docker_config(host)
        if not creds:
            # Try dockerhub-style keying (some helpers store with scheme)
            creds = (
                get_registry_creds_from_docker_config(f"https://{host}")
                or get_registry_creds_from_docker_config(f"http://{host}")
            )

        if creds and verbose:
            eprint(f"[i] {host}: acquiring bearer token for {repo_path} using docker-stored creds")

        if creds:
            tok = session.get(realm, params=token_params, auth=creds, timeout=30)
        else:
            tok = session.get(realm, params=token_params, timeout=30)

        if tok.status_code != 200:
            if verbose:
                eprint(f"[!] Token request failed ({tok.status_code}) for {host}/{repo_path}: {tok.text[:200]}")
            return None
        data = tok.json()
        return data.get("token") or data.get("access_token")
    except Exception as e:
        if verbose:
            eprint(f"[!] Token acquisition error for {host}/{repo_path}: {e}")
        return None


def acr_list_tags_detailed(image_ref: str, verbose: bool) -> List[Dict]:
    """
    Azure Container Registry tag listing via ACR data-plane HTTP API:

      https://<registry>.azurecr.io/acr/v1/<repository>/_tags?orderby=timedesc&n=...

    Returns list of dicts: {"name": <tag>, "lastUpdateTime": <iso str>}
    Auth: uses registry Bearer token derived from existing Docker login credentials (docker/az acr login).
    """
    image_no_tag, _ = split_image_and_tag(image_ref)
    host, repo = image_no_tag.split("/", 1)

    session = requests.Session()

    token = _registry_bearer_token(host, repo, "pull", session=session, verbose=verbose)
    headers = {}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    else:
        if verbose:
            eprint(f"[!] ACR: could not obtain bearer token for {host}/{repo}. Tag listing likely to fail.")
    # Pull ACR tag listing (best-effort pagination)
    out: List[Dict] = []
    n = 100
    last = None

    while True:
        params = {"orderby": "timedesc", "n": str(n)}
        if last:
            params["last"] = last
        url = f"https://{host}/acr/v1/{repo}/_tags"
        resp = session.get(url, headers=headers, params=params, timeout=30)
        if resp.status_code != 200:
            if verbose:
                eprint(f"[!] ACR tag listing failed ({resp.status_code}) for {host}/{repo}: {resp.text[:200]}")
            break
        data = resp.json()
        tags = (data.get("tags") or [])
        if not tags:
            break
        for t in tags:
            name = t.get("name")
            lut = t.get("lastUpdateTime") or t.get("last_update_time") or t.get("createdTime")
            if name:
                out.append({"name": name, "lastUpdateTime": lut})
        # Pagination: ACR returns "link" object sometimes, and "last" token for next call
        last = data.get("last")
        if not last:
            # If no cursor, stop.
            break
    return out


def acr_list_tags(image_ref: str, verbose: bool) -> List[str]:
    return [t["name"] for t in acr_list_tags_detailed(image_ref, verbose=verbose)]

# =========================
# Scanning logic
# =========================

def normalize_tar_path(member_name: str) -> str:
    s = member_name.lstrip("./")
    if not s.startswith("/"):
        s = "/" + s
    return s

def get_extension(path: str) -> str:
    base = path.rsplit("/", 1)[-1]
    if "." not in base or base.startswith(".") and base.count(".") == 1:
        return ""
    return "." + base.rsplit(".", 1)[-1].lower()

def is_text_like(raw: bytes) -> bool:
    if not raw:
        return False
    # NUL byte is a strong binary indicator
    if b"\x00" in raw:
        return False
    # Printable ratio heuristic
    printable = 0
    for b in raw[:8192]:
        if b in (9, 10, 13) or 32 <= b <= 126:
            printable += 1
    ratio = printable / max(1, min(len(raw), 8192))
    return ratio >= 0.85

def build_mode_filters(args) -> Dict:
    """
    Returns compiled excludes / allowlists based on mode + user overrides.
    """
    exclude_path = []
    exclude_ext = set()
    content_allow = set()
    deep_roots = list(args.deep_root or [])
    system_roots = list(args.system_root or [])

    if args.mode == "linux-container":
        exclude_path = [re.compile(p) for p in LINUX_CONTAINER_EXCLUDE_PATH_REGEX]
        exclude_ext |= set(LINUX_CONTAINER_EXCLUDE_EXTENSIONS)
        content_allow |= set(LINUX_CONTAINER_CONTENT_EXT_ALLOWLIST)
        deep_roots.extend(list(LINUX_CONTAINER_DEEP_ROOTS))
        system_roots.extend(list(LINUX_CONTAINER_SYSTEM_ROOTS))

        # Allow secrets paths under /run despite broad /run exclusion above
        # We'll implement as a "negative" check later: deep roots win.
    else:
        exclude_path = []
        exclude_ext = set()
        content_allow = set()

    # Apply user overrides
    for p in (args.exclude_path_regex or []):
        try:
            exclude_path.append(re.compile(p))
        except re.error as e:
            eprint(f"[!] Invalid --exclude-path-regex '{p}': {e}")

    for e in (args.exclude_ext or []):
        if not e:
            continue
        if not e.startswith("."):
            e = "." + e
        exclude_ext.add(e.lower())

    for e in (args.content_ext or []):
        if not e:
            continue
        if not e.startswith("."):
            e = "." + e
        content_allow.add(e.lower())

    # Dedup roots and ensure absolute
    deep_roots = sorted({r if r.startswith("/") else "/" + r for r in deep_roots})
    system_roots = sorted({r if r.startswith("/") else "/" + r for r in system_roots})

    return {
        "exclude_path": exclude_path,
        "exclude_ext": exclude_ext,
        "content_allow": content_allow,
        "deep_roots": tuple(deep_roots),
        "system_roots": tuple(system_roots),
    }

def should_skip_path(abs_path: str, filters: Dict) -> bool:
    # Deep roots always override excludes (e.g., /run/secrets)
    for dr in filters["deep_roots"]:
        if abs_path.startswith(dr.rstrip("/") + "/") or abs_path == dr:
            return False
    for rx in filters["exclude_path"]:
        if rx.search(abs_path):
            return True
    return False

def is_in_roots(abs_path: str, roots: Tuple[str, ...]) -> bool:
    for r in roots:
        if abs_path == r or abs_path.startswith(r.rstrip("/") + "/"):
            return True
    return False

def extract_env_vars(container) -> List[str]:
    """Return raw environment variables (no redaction)."""
    try:
        # Ensure attrs are populated
        try:
            container.reload()
        except Exception:
            pass
        env_vars = container.attrs.get("Config", {}).get("Env") or []
        # Ensure list[str]
        return [str(x) for x in env_vars]
    except Exception:
        return []

def scan_container_filesystem(container, cfg: RegexConfig, max_bytes: int, scan_paths: List[str], verbose: bool, filters: Dict, max_content_file_bytes: int) -> Dict:
    findings = {"filename_hits": [], "content_hits": []}
    tmp_tar_path = None

    # Export per path; Docker SDK wants exact path in container
    for root in scan_paths:
        try:
            stream, _ = container.get_archive(root)
            with tempfile.NamedTemporaryFile(delete=False) as tmp_tar:
                for chunk in stream:
                    tmp_tar.write(chunk)
                tmp_tar_path = tmp_tar.name

            with tarfile.open(tmp_tar_path) as tar:
                for member in tar.getmembers():
                    if not member.isfile():
                        continue

                    name = member.name
                    abs_path = normalize_tar_path(name)
                    if should_skip_path(abs_path, filters):
                        continue

                    # Filename match
                    if any(p.search(abs_path) for p in cfg.filename_patterns):
                        findings["filename_hits"].append(abs_path)

                    # Content match (read up to max_bytes) with linux-container gating
                    ext = get_extension(abs_path)
                    in_system = is_in_roots(abs_path, filters["system_roots"]) and not abs_path.startswith("/usr/local/")
                    in_etc = abs_path.startswith("/etc/")
                    # Determine if we should attempt content scanning
                    allow_by_ext = (not filters["content_allow"]) or (ext in filters["content_allow"]) or (ext == "")
                    if filters["exclude_ext"] and ext in filters["exclude_ext"]:
                        allow_by_ext = False
                    # System trees: only scan config-like files (allowlist ext) or small no-ext text
                    if in_system and ext not in filters["content_allow"] and ext != "":
                        allow_by_ext = False
                    # Skip huge files for content scanning
                    if member.size and member.size > max_content_file_bytes:
                        allow_by_ext = False
                    if not allow_by_ext:
                        continue

                    try:
                        f = tar.extractfile(member)
                        if not f:
                            continue
                        raw = f.read(max_bytes)
                        # Fast skip if empty
                        if not raw:
                            continue
                        # Skip binary-like blobs (reduces base-image noise)
                        if not is_text_like(raw):
                            continue
                        text = raw.decode("utf-8", errors="ignore")

                        # Identify the specific lines that triggered matches
                        matches = []
                        max_lines_per_file = 50
                        max_chars_per_line = 2000
                        for lineno, line in enumerate(text.splitlines(), start=1):
                            if not line:
                                continue
                            # Suppress obvious benign contexts for generic assignment hits
                            benign = bool(BENIGN_CONTEXT_RE.search(line))
                            for p in cfg.content_patterns:
                                if p.search(line):
                                    if benign and p.pattern.startswith('(?i)\\b'):
                                        continue
                                    # record the first occurrence per (pattern, line) to reduce noise
                                    matches.append({
                                        "pattern": p.pattern,
                                        "line_number": lineno,
                                        "line": line[:max_chars_per_line],
                                    })
                                    if len(matches) >= max_lines_per_file:
                                        break
                            if len(matches) >= max_lines_per_file:
                                break

                        if matches:
                            # Keep backward-compatible summary too
                            matched_patterns = sorted({m["pattern"] for m in matches})
                            findings["content_hits"].append({
                                "file": abs_path,
                                "matched_patterns": matched_patterns,
                                "matches": matches,
                            })
                    except Exception:
                        continue
        except Exception as e:
            if verbose:
                eprint(f"[!] Failed to export '{root}' from container: {e}")
        finally:
            if tmp_tar_path and os.path.exists(tmp_tar_path):
                try:
                    os.remove(tmp_tar_path)
                except Exception:
                    pass
            tmp_tar_path = None

    # Deduplicate for cleanliness
    findings["filename_hits"] = sorted(set(findings["filename_hits"]))
    # content_hits keep duplicates possible; dedupe by file+matched_patterns (merge matches)
    seen = {}
    for hit in findings["content_hits"]:
        key = (hit.get("file"), tuple(sorted(hit.get("matched_patterns", []))))
        if key not in seen:
            seen[key] = hit
        else:
            # Merge match entries if present
            existing = seen[key]
            if "matches" in hit:
                ex = existing.setdefault("matches", [])
                # merge unique by (pattern, line_number, line)
                ex_set = {(m.get("pattern"), m.get("line_number"), m.get("line")) for m in ex}
                for m in hit.get("matches", []):
                    t = (m.get("pattern"), m.get("line_number"), m.get("line"))
                    if t not in ex_set:
                        ex.append(m)
                        ex_set.add(t)
                # refresh summary patterns
                existing["matched_patterns"] = sorted({m.get("pattern") for m in ex if m.get("pattern")})
    findings["content_hits"] = list(seen.values())

    return findings

def scan_image_tag(docker_client: docker.DockerClient, image_tag: str, cfg: RegexConfig, max_bytes: int,
                   scan_paths: List[str], verbose: bool, filters: Dict, max_content_file_bytes: int) -> Dict:
    """
    Returns a report row.
    """
    row = {
        "image": image_tag,
        "status": "ok",
        "environment_variables": [],
        "interesting_environment_variables": [],
        "filename_hits": [],
        "content_hits": [],
        "counts": {"env": 0, "interesting_env": 0, "filename_hits": 0, "content_hits": 0},
        "errors": [],
    }
    container = None
    try:
        if isinstance(image_tag, list):
            # Defensive: should never happen, but normalize
            image_tag = image_tag[0] if image_tag else ""
        if verbose:
            eprint(f"[>] Pulling {image_tag}")
        image = docker_client.images.pull(image_tag)

        # Create container (no start needed for get_archive)
        container = docker_client.containers.create(image.id, command="sleep 1d")

        row["environment_variables"] = extract_env_vars(container)
        row["interesting_environment_variables"] = filter_interesting_env(row["environment_variables"])
        fs = scan_container_filesystem(container, cfg, max_bytes=max_bytes, scan_paths=scan_paths, verbose=verbose,
                                      filters=filters, max_content_file_bytes=max_content_file_bytes)
        row["filename_hits"] = fs["filename_hits"]
        row["content_hits"] = fs["content_hits"]
        row["counts"] = {
            "environment_variables": len(row["environment_variables"]),
            "interesting_env": len(row.get("interesting_environment_variables") or []),
            "filename_hits": len(row["filename_hits"]),
            "content_hits": len(row["content_hits"]),
        }
        return row
    except Exception as e:
        row["status"] = "error"
        row["errors"].append(str(e))
        return row
    finally:
        if container is not None:
            try:
                container.remove(force=True)
            except Exception:
                pass

# =========================
# Main
# =========================

def enumerate_tags_for_image(image_ref: str, registry: str, verbose: bool) -> List[str]:
    """Enumerate tags for an image reference (no tag => list tags; explicit tag => that tag only)."""
    image_no_tag, maybe_tag = split_image_and_tag(image_ref)
    if maybe_tag:
        return [maybe_tag]

    session = requests.Session()

    if registry == "dockerhub":
        # 1) Try anonymous token (works for public repos)
        tags = dockerhub_list_tags(image_no_tag, session=session, verbose=verbose, auth=None)
        if tags:
            return tags

        # 2) Try with whatever Docker already has stored for Docker Hub (docker login / hub-tool / credential helper)
        creds = (
            get_registry_creds_from_docker_config("https://index.docker.io/v1/")
            or get_registry_creds_from_docker_config("index.docker.io")
            or get_registry_creds_from_docker_config("registry-1.docker.io")
            or get_registry_creds_from_docker_config("docker.io")
        )
        if not creds:
            if verbose:
                eprint("[!] Docker Hub: no stored docker credentials found; cannot enumerate tags for private repos.")
            return []

        return dockerhub_list_tags(image_no_tag, session=session, verbose=verbose, auth=creds)

    if registry == "acr":
        # Use ACR HTTP API; relies on existing Docker login creds/credential helper.
        return acr_list_tags(image_no_tag, verbose=verbose)

    if registry == "ecr":
        return ecr_list_tags(image_no_tag, verbose=verbose)

    if registry == "gcr":
        return gcr_list_tags(image_no_tag, verbose=verbose)

    if verbose:
        eprint(f"[!] Tag listing not implemented for registry type '{registry}'.")
    return []


def build_image_tags(image_ref: str, tags: List[str]) -> List[str]:
    image_no_tag, maybe_tag = split_image_and_tag(image_ref)
    if maybe_tag:
        return [f"{image_no_tag}:{maybe_tag}"]
    return [f"{image_no_tag}:{t}" for t in tags]

def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="TagTrawler - Docker image secrets scanner")
    p.add_argument("-r", "--remote", nargs="+", help="Image refs (Docker Hub: namespace/repo, ECR: <acct>.dkr.ecr.<region>.amazonaws.com/repo, GCR: gcr.io/project/image, etc.)")
    p.add_argument("--all-tags", action="store_true", help="Enumerate and scan all tags for each --remote ref")
    p.add_argument("--registry", choices=["dockerhub", "ecr", "gcr", "acr", "other"], help="Override registry type (auto-detect by default)")
    p.add_argument("--threads", type=int, default=5, help="Parallel tag scanning threads")
    p.add_argument("--regex-config", help="JSON or YAML file with filename_patterns and content_patterns")
    p.add_argument("--max-bytes", type=int, default=65536, help="Max bytes to read from each file for content scanning")
    p.add_argument("--scan-path", action="append", default=[], help="Restrict export/scan to specific path(s) in container. Repeatable. Default: /")
    p.add_argument("-o", "--output", default=f"TagTrawler_report_{utc_now_compact()}.json", help="Output JSON report path")
    p.add_argument("--verbose", action="store_true", help="Verbose logs")
    p.add_argument("--mode", choices=["linux-container", "default"], default="default", help="Noise-reduction profile. linux-container reduces base-image chaff while preserving app paths.")
    p.add_argument("--exclude-path-regex", action="append", default=[], help="Regex for absolute paths to skip entirely. Repeatable.")
    p.add_argument("--exclude-ext", action="append", default=[], help="File extensions (e.g., .png) to skip content scanning for. Repeatable.")
    p.add_argument("--content-ext", action="append", default=[], help="Allowlist extensions for content scanning. Repeatable.")
    p.add_argument("--max-content-file-bytes", type=int, default=524288, help="Skip content scanning for files larger than this (default 512KB).")
    p.add_argument("--deep-root", action="append", default=[], help="Absolute path prefix to always deep-scan (content scanning rules still apply). Repeatable.")
    p.add_argument("--system-root", action="append", default=[], help="Absolute path prefix treated as system tree (more restrictive content scan). Repeatable.")

    p.add_argument("--specific-tag", help="Scan only this tag for each image (overrides any tag in -r and skips tag enumeration). Works with or without --all-tags.")
    return p.parse_args()

def main() -> int:
    args = parse_args()
    # Back-compat: older builds used action="append" for --specific-tag.
    if isinstance(getattr(args, "specific_tag", None), list):
        if len(args.specific_tag) == 0:
            args.specific_tag = None
        elif len(args.specific_tag) == 1:
            args.specific_tag = args.specific_tag[0]
        else:
            raise SystemExit("[!] --specific-tag may be provided only once.")
    cfg = load_regex_config(args.regex_config)
    filters = build_mode_filters(args)
    scan_paths = args.scan_path if args.scan_path else DEFAULT_SCAN_PATHS

    if not args.remote:
        eprint("[!] No --remote images specified. (Local image enumeration not enabled in this refactor.)")
        return 2

    docker_client = docker.from_env()
    rows: List[Dict] = []
    futures = []
    lock = threading.Lock()

    def submit_scan(tag: str):
        return scan_image_tag(
            docker_client, tag,
            cfg=cfg,
            max_bytes=args.max_bytes,
            scan_paths=scan_paths,
            verbose=args.verbose,
            filters=filters,
            max_content_file_bytes=args.max_content_file_bytes,
        )

    to_scan: List[str] = []
    for ref in args.remote:
        reg = args.registry or detect_registry(ref)
        if args.verbose:
            eprint(f"[i] Registry for {ref}: {reg}")

        # --specific-tag is a direct override: scan exactly that tag for each provided repo,
        # regardless of --all-tags. This prevents falling back to ':latest' unintentionally.
        if args.specific_tag:
            image_no_tag, explicit = split_image_and_tag(ref)
            if explicit and explicit != args.specific_tag and args.verbose:
                eprint(f"[!] Both an explicit tag '{explicit}' and --specific-tag '{args.specific_tag}' were provided; using --specific-tag.")
            to_scan.append(f"{image_no_tag}:{args.specific_tag}")
            continue

        if args.all_tags:
            tags = enumerate_tags_for_image(ref, registry=reg, verbose=args.verbose)

            if not tags:
                eprint(f"[!] No tags found for {ref}. Try: (a) verify repo exists, (b) ensure Docker is logged in, (c) specify explicit tag, or (d) use --specific-tag <tag>.")
                continue

            if args.verbose:
                eprint(f"[i] {ref}: scanning {len(tags)} tag(s)")

            to_scan.extend(build_image_tags(ref, tags))
        else:
            # If no explicit tag, default to latest
            image_no_tag, maybe_tag = split_image_and_tag(ref)
            tag = maybe_tag or "latest"
            to_scan.append(f"{image_no_tag}:{tag}")

    if not to_scan:
        eprint("[!] Nothing to scan.")
        with open(args.output, "w", encoding="utf-8") as f:
            json.dump([], f, indent=2)
        return 1

    if args.verbose:
        eprint(f"[i] Total image:tag targets: {len(to_scan)}")

    with cf.ThreadPoolExecutor(max_workers=max(1, args.threads)) as ex:
        try:
            for image_tag in to_scan:
                futures.append(ex.submit(submit_scan, image_tag))

            for fut in cf.as_completed(futures):
                row = fut.result()
                with lock:
                    rows.append(row)

                if args.verbose:
                    eprint(f"[i] Done: {row['image']} status={row['status']} hits={row['counts']}")
        except KeyboardInterrupt:
            eprint("[!] Interrupted by user. Writing partial results...")
            # Best-effort cancel pending futures
            for f in futures:
                try:
                    f.cancel()
                except Exception:
                    pass

    # Sort output for stable diffs (by image name)
    rows.sort(key=lambda r: r.get("image", ""))

    # "table-like" JSON: list of rows with normalized keys
    with open(args.output, "w", encoding="utf-8") as f:
        json.dump(rows, f, indent=2)

    eprint(f"[+] Report saved to {args.output} ({len(rows)} rows)")
    return 0

if __name__ == "__main__":
    raise SystemExit(main())