auto_feed_discovery.yml

name: Auto Feed Discovery

on:
  schedule:
    # Sonntag mit Backup-Slots gegen GitHub-Skips bei hoher Last.
    # Krumme Minuten meiden top-of-hour congestion. Idempotenz-Guard im
    # ersten Step skippt Folge-Trigger, wenn heute bereits erfolgreich gelaufen.
    - cron: '37 4 * * 0'   # 04:37 UTC – primärer Slot (nach cve_to_ip_mapper um 04:00)
    - cron: '23 7 * * 0'   # 07:23 UTC – Backup #1 falls primärer Slot von GitHub übersprungen
    - cron: '47 11 * * 0'  # 11:47 UTC – Backup #2 als letzte Sicherheit
  workflow_dispatch:

env:
  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true

permissions:
  contents: write
  actions: read
  issues: none
  packages: none
  pull-requests: none
  security-events: none

concurrency:
  group: netshield-seen-db-writers
  cancel-in-progress: false

jobs:
  discover:
    runs-on: ubuntu-latest
    timeout-minutes: 45  # erhöht von 30 – bei Rate-Limit-Wartezeiten (60s) kann 30min knapp werden

    steps:
      # Idempotenz-Guard: Bei mehreren Cron-Slots (4:37, 7:23, 11:47 UTC) verhindert
      # dieser Step, dass Backup-Trigger laufen, wenn der primäre Slot bereits
      # erfolgreich war. Bei manuellem Dispatch wird der Guard übersprungen
      # (github.event_name != 'schedule'), sodass forcierte Runs immer durchgehen.
      - name: Skip if already succeeded today
        id: idempotency
        if: github.event_name == 'schedule'
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: |
          today=$(date -u +%Y-%m-%d)
          success_count=$(gh run list \
            --repo "${{ github.repository }}" \
            --workflow=auto_feed_discovery.yml \
            --status=success \
            --created="$today" \
            --limit=10 \
            --json databaseId \
            --jq 'length')
          if [ "$success_count" -gt 0 ]; then
            echo "already_ran=true" >> "$GITHUB_OUTPUT"
            echo "Heute bereits $success_count erfolgreiche Run(s) – Skip dieses Backup-Triggers"
          else
            echo "already_ran=false" >> "$GITHUB_OUTPUT"
            echo "Noch kein erfolgreicher Run heute – fortfahren"
          fi

      - name: Checkout Repository
        if: steps.idempotency.outputs.already_ran != 'true'
        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
        with:
          persist-credentials: true

      - name: Restore seen_db Cache
        if: steps.idempotency.outputs.already_ran != 'true'
        uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
        with:
          path: seen_db.json
          # Eigener afd-Prefix verhindert dass auto_feed_discovery den
          # Combined-Cache (v2-Prefix) überschreibt. restore-keys lesen
          # zuerst den letzten eigenen afd-Stand, dann den Combined-Stand.
          key: netshield-seen-db-afd-save-${{ github.run_id }}
          restore-keys: |
            netshield-seen-db-afd-save-
            netshield-seen-db-afd-
            netshield-seen-db-v2-
            netshield-seen-db-v1-
            netshield-seen-db-

      - name: Discover, Validate and Ingest IP Feeds
        if: steps.idempotency.outputs.already_ran != 'true'
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: |
          # FIX UNBUFFERED: -u erzwingt ungepufferten stdout. Ohne das block-puffert
          # CPython, weil GitHub Actions kein TTY ist – bei einem Job-Cancel
          # (SIGTERM/Timeout) gehen alle bisherigen prints verloren, und im Log
          # steht nichts ausser dem Heredoc-Echo. Mit -u streamen Logs live und
          # zeigen genau, wo das Skript hing.
          python3 -u << 'EOF'
          import sys as _sys; _sys.path.insert(0, "scripts")
          from netshield_common import (
              load_whitelist, load_fp_set, is_in_fp_set,
              is_valid_public_ipv4, is_valid_public_cidr,
              is_protected_entry, is_whitelisted,
              parse_entries as _parse_entries, calculate_confidence,
              safe_get_date, parse_date, sort_ips, write_ip_list,
              write_json_atomic, write_text_atomic,
              fetch_url, check_local_feed_age,
              IPV4_RE, CIDR_RE, TIMESTAMP_RE,
          )

          # Init: Whitelist + FP-Set laden
          load_whitelist()
          load_fp_set()

          import urllib.request
          import urllib.error
          import urllib.parse
          import json
          import re
          import os
          import sys
          import time
          import ipaddress
          from datetime import datetime, timezone, timedelta
          from collections import defaultdict
          from concurrent.futures import ThreadPoolExecutor


          # FIX PERF-TIMEOUT: Globale Timeout-Sicherheit.
          # Der Workflow hat ein 45-Minuten-Limit. Bei unerwartet langsamen
          # externen Quellen oder API-Delays soll das Skript vor dem Timeout
          # sauber exiten und den Report committen (if: always()).
          _JOB_START = time.time()
          _MAX_RUNTIME_SECONDS = 40 * 60  # 40 Minuten = harter Cutoff
          def _check_timeout(label=""):
              elapsed = time.time() - _JOB_START
              if elapsed > _MAX_RUNTIME_SECONDS:
                  print(f"::warning file=auto_feed_discovery.yml::"
                        f"Timeout-Sicherheit ausgelöst ({label}): "
                        f"{elapsed/60:.1f}min elapsed, breche ab um "
                        f"sauberen Exit vor dem 45min-Limit zu gewährleisten.")
                  return True
              return False


          now     = datetime.now(timezone.utc)
          now_str = now.strftime("%Y-%m-%d %H:%M UTC")
          now_day = now.strftime("%Y-%m-%d")

          GH_TOKEN   = os.environ.get("GH_TOKEN", "")
          # IPV4_RE wird aus netshield_common importiert (strikte Variante mit
          # Lookbehind/-ahead auf '.', filtert Versions-Strings wie '1.2.3.4.5'
          # zuverlässig). Frühere lokale Re-Definition war eine schwächere
          # Variante (nur \d-Lookbehind) und liess Versions-IPs durch.
          READ_LIMIT = 5 * 1024 * 1024   # 5 MB Sample pro Feed

          # ── Bekannte Repos – NICHT erneut aufnehmen ───────────────────────
          KNOWN_REPOS = {
              "firehol/blocklist-ipsets", "stamparm/ipsum", "stamparm/maltrail",
              "montysecurity/C2-Tracker", "drb-ra/C2IntelFeeds",
              "Gi7w0rm/CobaltStrikeC2Tracker", "borestad/blocklist-abuseipdb",
              "elliotwutingfeng/ThreatFox-IOC-IPs", "elliotwutingfeng/ESET-APR-IPs",
              "elliotwutingfeng/Spamhaus-DNSBL-IPs", "romainmarcoux/malicious-ip",
              "romainmarcoux/malicious-outgoing-ip", "duggytuxy/malicious_ip_addresses",
              "duggytuxy/Data-Shield_IPv4_Blocklist", "ShadowWhisperer/IPs",
              "GridinSoft/IP-Blocklist", "T145/black-mirror", "LittleJake/ip-blacklist",
              "actuallymentor/bluetack-ip-blacklist-generator", "sefinek/Malicious-IP-Addresses",
              "scriptzteam/AbuseIPDB-BlackList", "axllent/iplists",
              "mrlooker/Suspicious-IPs", "MagicTeaMC/bad-ips",
              "ddrimus/http-threat-blocklist", "BlacKSnowDot0/packetsdatabase-db",
              "FreakuencyFive/Public-IP-ThreatFeed", "amitambekar510/Malicious-IP-Threat-List",
              "dolutech/blacklist-dolutech", "ZEROF/ipextractor",
              "coyote-nl/blocklist", "bitwire-it/ipblocklist",
              "NETMOUNTAINS/Curated-IP-Blocklist", "4IP-Solutions/threat-feeds",
              "cybersecurity-cyna/Malicious_IP", "lula73/bot-detector",
              "CriticalPathSecurity/Public-Intelligence-Feeds",
              "SecOps-Institute/Tor-IP-Addresses", "Enkidu-6/tor-relay-lists",
              "yuexuana521/honeypot-blocklist", "EdanWong/ip_list",
              "nixbear/malicious_ips", "ufukart/Blacklist", "f3csystems/BlockList_IP",
              "IT3ngineer/FortigateBlockList", "florentvinai/bad-ips-on-my-vps",
              "Y3ll0w/CrowdSec-CAPI-Decisions", "cenk/trcert-malware",
              "Tizian-Maxime-Weigt/L7-HTTP-DDoS-Flood-IP-Signature-IP-List",
              "Ultimate-Hosts-Blacklist/Ultimate.Hosts.Blacklist",
              "tushroy/bdix-isp-blocks", "jgamblin/Shodan-Malware-Scanning-IPs",
          }

          # ── Ausschluss-Keywords ───────────────────────────────────────────
          EXCLUDE_KEYWORDS = {
              "geoip", "geo-ip", "geo_ip", "country", "countries", "geolocation",
              "vpn", "datacenter", "cdn", "cloud-ip", "cloud-range",
              "alienvault", "otx", "ads", "adblock", "adguard", "hosts",
              "dns-blocklist", "domain", "pihole",
          }

          # ── Dateinamen-Muster für IP-Listen ───────────────────────────────
          # Erlaubt optional .gz-Suffix – fetch_url dekomprimiert transparent.
          IP_FILE_PATTERNS = re.compile(
              r'(?i)(?:'
              r'ip[s_-]?(?:black|block|ban|threat|malicious|bad|abuse|deny|reputation|ioc)?list|'
              r'blacklist|blocklist|denylist|banlist|'
              r'threat[-_]?feed|threatintel|threat[-_]?intel|'
              r'malicious|malicious[-_]?ips?|bad[-_]?ips?|'
              r'suspicious[-_]?ips?|'
              r'ipblock|ioc|iocs|c2|botnet|scanner|scanners|'
              r'bruteforce|ssh|rdp|attack|attackers|'
              r'tor|honeypot|reputation'
              r')\.(?:txt|csv|ipset|netset|list|conf|sh|dat)(?:\.gz)?$'
          )

          # ── Whitelist-CIDRs (False-Positive-Schutz) ──────────────────────
          # FIX SSOT4: whitelist.json als Single Source of Truth laden.
          # Ersetzt hardcoded FP_CIDRS, DNS_WHITELIST und PROTECTED_CIDRS.
          # Synchron mit update_combined_blacklist, false_positive_checker
          # und community_ip_report.
          try:
              with open(".github/workflows/whitelist.json", encoding="utf-8") as _wl_f:
                  _WHITELIST_ENTRIES = json.load(_wl_f)["entries"]
          except Exception as _wl_err:
              _msg = f"whitelist.json nicht ladbar: {_wl_err} – Auto Feed Discovery abgebrochen"
              print(f"::error file=auto_feed_discovery.yml::{_msg}")
              print(f"FEHLER: {_msg}")
              sys.exit(1)

          fp_nets = []
          for _wl_entry in _WHITELIST_ENTRIES:
              try:
                  fp_nets.append(ipaddress.ip_network(_wl_entry, strict=False))
              except Exception as _suppressed:
                  print(f"WARN: suppressed Exception: {_suppressed}", file=sys.stderr)

          # Protected-Nets für is_valid_public_ipv4: identisch mit fp_nets
          # (whitelist.json enthält bereits alle privaten Ranges, DNS, CDN etc.)
          protected_nets = fp_nets

          # ── Hilfsfunktionen ───────────────────────────────────────────────
          def gh_api(path, token=GH_TOKEN, retries=3):
              # FIX RATE-LIMIT: bei 403/429 X-RateLimit-Reset auswerten. Wenn
              # Reset in ≤ MAX_WAIT_SECONDS liegt, einmal warten – sonst sofort
              # raus. Frühere Variante (30 min wait) führte mit parallelem
              # REFRESH zum Job-Timeout-Tod, weil mehrere Worker gleichzeitig
              # sleepten. Pragmatik: einen Call opfern ist billiger als ganzer
              # Run failt; der nächste Sonntags-Run holt's nach.
              # FIX 5xx-RETRY: 500/502/503/504 vorher → sofortiges None.
              # Jetzt exponentielles Backoff (5s, 10s, 20s).
              url = f"https://api.github.com{path}"
              headers = {"Accept": "application/vnd.github+json",
                         "X-GitHub-Api-Version": "2022-11-28"}
              if token:
                  headers["Authorization"] = f"Bearer {token}"

              TRANSIENT_CODES = {500, 502, 503, 504}
              MAX_WAIT_SECONDS = 90   # Hard-Cap: niemals länger als 90s warten

              for attempt in range(retries):
                  try:
                      req = urllib.request.Request(url, headers=headers)
                      with urllib.request.urlopen(req, timeout=20) as r:
                          return json.loads(r.read().decode("utf-8"))
                  except urllib.error.HTTPError as e:
                      if e.code in (403, 429):
                          reset_str = ""
                          try:
                              reset_str = e.headers.get("X-RateLimit-Reset", "") if e.headers else ""
                          except Exception:
                              pass
                          wait = None
                          try:
                              if reset_str:
                                  candidate = int(reset_str) - int(time.time()) + 5
                                  if 0 < candidate <= MAX_WAIT_SECONDS:
                                      wait = candidate
                          except (ValueError, TypeError):
                              pass
                          if wait is not None and attempt < retries - 1:
                              print(f"  Rate-Limit (HTTP {e.code}) – warte {wait}s bis Reset...")
                              time.sleep(wait)
                              continue
                          # Reset zu weit weg, fehlt, oder Retries erschöpft – skip
                          print(f"  Rate-Limit (HTTP {e.code}) – skip {path} "
                                f"(Reset >{MAX_WAIT_SECONDS}s oder fehlt)")
                          return None
                      elif e.code == 422:
                          return None
                      elif e.code in TRANSIENT_CODES:
                          if attempt < retries - 1:
                              sleep_s = 5 * (2 ** attempt)
                              print(f"  HTTP {e.code} (transient) {path} – warte {sleep_s}s, "
                                    f"Retry {attempt+1}/{retries}")
                              time.sleep(sleep_s)
                              continue
                          print(f"  HTTP {e.code} – Retries erschöpft für {path}")
                          return None
                      else:
                          print(f"  HTTP {e.code} für {path}")
                          return None
                  except Exception as ex:
                      if attempt < retries - 1:
                          time.sleep(5)
                      else:
                          print(f"  Fehler API {path}: {ex}")
                          return None

          def fetch_raw(url, limit=READ_LIMIT, timeout=25):
              # FIX SEC: Nutzt fetch_url aus netshield_common – enthält SSRF-Schutz
              # (Schema-/Host-Validierung, auch bei Redirects). Ohne diesen Schutz
              # könnte eine aus GitHub-Suchergebnissen stammende URL auf
              # 169.254.169.254 / localhost / RFC1918 zeigen.
              return fetch_url(url, timeout=timeout, retries=1,
                               user_agent="NETSHIELD-AutoDiscovery/2.0",
                               read_limit=limit)

          # is_valid_public_ipv4() → importiert aus netshield_common
          def extract_ips(text):
              # FIX BUG-WF4-IPV6: Vorher eigene IPV4_RE-Schleife mit
              # is_valid_public_ipv4 – ohne den IPv6-Token-Schutz von
              # parse_entries. '::ffff:1.2.3.4' in einem Discovery-
              # Kandidaten-File haette Phantom-IPv4-Eintraege erzeugt
              # (gleicher BUG-IPV6-MAPPED wie in netshield_common).
              # Jetzt: parse_entries() nutzen, dann CIDRs herausfiltern
              # damit der Vertrag der Funktion (nur Plain-IPs) erhalten
              # bleibt – fp_rate / hq_ref_ips-Overlap rechnen mit
              # IP-Strings, ein CIDR wuerde dort ValueError ausloesen
              # bzw. die Set-Operationen verzerren. Whitelist-Filter
              # NICHT aktiv (use_protected_check=False), weil Discovery
              # die Liste nur evaluiert.
              return {e for e in _parse_entries(text) if "/" not in e}

          def fp_rate(ips, sample_size=200):
              # FIX SAMPLE-BIAS: random.sample statt list(ips)[:200]. Set-Iteration
              # ist zwar nicht streng sortiert, aber die ersten N Elemente korrelieren
              # in der Praxis stark (Hash-Bucketing nach IP-Bytes) → Stichprobe trifft
              # oft denselben /24-Block, was die FP-Rate verzerrt.
              # Seed=42 für deterministische Reproduzierbarkeit zwischen Runs:
              # Borderline-Feeds bei ~5% FP-Rate flackern sonst zwischen accept/reject.
              import random as _random
              ips_list = list(ips)
              if not ips_list:
                  return 1.0
              if len(ips_list) > sample_size:
                  sample = _random.Random(42).sample(ips_list, sample_size)
              else:
                  sample = ips_list
              hits = 0
              for ip_str in sample:
                  try:
                      addr = ipaddress.ip_address(ip_str)
                      if any(addr in net for net in fp_nets):
                          hits += 1
                  except ValueError as _suppressed:
                      print(f"WARN: suppressed ValueError: {_suppressed}", file=sys.stderr)
              return hits / len(sample)

          def add_repo_candidate(candidate_repos, repo, source_label):
              if not repo:
                  return
              full_name = repo.get("full_name")
              if not full_name or full_name in KNOWN_REPOS:
                  return
              combined = " ".join([
                  full_name.lower(),
                  " ".join(repo.get("topics", [])),
                  (repo.get("description") or "").lower()
              ])
              if any(kw in combined for kw in EXCLUDE_KEYWORDS):
                  return
              existing = candidate_repos.get(full_name, {})
              sources = set(existing.get("sources", []))
              sources.add(source_label)
              candidate_repos[full_name] = {
                  "stars":       repo.get("stargazers_count", existing.get("stars", 0)),
                  "forks":       repo.get("forks_count",      existing.get("forks", 0)),
                  "updated":     repo.get("updated_at",       existing.get("updated", "")),
                  "pushed":      repo.get("pushed_at",        existing.get("pushed", "")),
                  "description": repo.get("description",      existing.get("description", "")),
                  "topics":      repo.get("topics",           existing.get("topics", [])),
                  "sources":     sorted(sources),
              }

          # ── Schritt 1: seen_db laden ──────────────────────────────────────
          # FIX BUG-CACHE-RESTORE: Identisch zu update_combined_blacklist.yml.
          # Wenn der Cache abgelaufen/leer ist und auto_feed_discovery (woechentlich
          # Sonntag) zuerst laeuft, wuerde er sonst mit db={} starten und nach
          # dem Lauf eine fast-leere seen_db zurueck in den Cache schreiben.
          # Combined wuerde dann beim naechsten Lauf diesen gekuerzten Cache
          # laden → komplette History (days_seen, first, hq_feed_names) verloren.
          # Backup-Restore aus seen_db_backup.json.gz (Repo-File) verhindert das.
          DB_FILE = "seen_db.json"
          BACKUP_FILE = "seen_db_backup.json.gz"
          db = {}
          if os.path.exists(DB_FILE):
              try:
                  with open(DB_FILE) as f:
                      db = json.load(f)
                  print(f"seen_db geladen: {len(db)} IPs")
              except Exception as _load_err:
                  print(f"WARN: seen_db.json korrupt ({_load_err}) – versuche Backup", file=sys.stderr)
                  db = {}
          if not db and os.path.exists(BACKUP_FILE):
              try:
                  import gzip
                  # FIX BUG-GZIP-BACKUP-LIMIT: identisch zu update_combined_blacklist.
                  # Ohne Pre-Size + Streaming-Limit wuerde eine zip-Bombe im
                  # Backup den Workflow per OOM kippen bevor _MIN_BACKUP_ENTRIES
                  # je greift. Layer 1: 200 MB komprimiert hard-cap.
                  # Layer 2: 2 GB Streaming-Read auf den dekomprimierten Text.
                  _BACKUP_COMPRESSED_LIMIT   = 200 * 1024 * 1024
                  _BACKUP_DECOMPRESSED_LIMIT = 2 * 1024 * 1024 * 1024
                  _bsize = os.path.getsize(BACKUP_FILE)
                  if _bsize > _BACKUP_COMPRESSED_LIMIT:
                      raise ValueError(
                          f"Backup zu gross: {_bsize / 1024 / 1024:.1f} MB "
                          f"> {_BACKUP_COMPRESSED_LIMIT / 1024 / 1024:.0f} MB "
                          f"(moegliche zip-Bombe oder Repo-Korruption)")
                  with gzip.open(BACKUP_FILE, "rt", encoding="utf-8") as _bf:
                      _raw = _bf.read(_BACKUP_DECOMPRESSED_LIMIT + 1)
                      if len(_raw) > _BACKUP_DECOMPRESSED_LIMIT:
                          raise ValueError(
                              f"Backup-Stream > "
                              f"{_BACKUP_DECOMPRESSED_LIMIT / 1024 / 1024:.0f} MB "
                              f"nach Dekomprimierung (zip-Bombe), verworfen")
                      db = json.loads(_raw)
                  # FIX BUG-BACKUP-GUARD: Backup-Plausibilitaet pruefen vor Restore.
                  # Ein winziges/leeres Backup wuerde sonst die Cache-Recovery
                  # mit kaputtem Stand zementieren - besser leer starten und
                  # neue Eintraege akkumulieren.
                  _MIN_BACKUP_ENTRIES = 10000
                  if not isinstance(db, dict) or len(db) < _MIN_BACKUP_ENTRIES:
                      print(f"WARN: Backup hat nur {len(db) if isinstance(db, dict) else 0} "
                            f"Eintraege (<{_MIN_BACKUP_ENTRIES}) - wird verworfen, starte neu.",
                            file=sys.stderr)
                      db = {}
                  else:
                      print(f"seen_db aus Backup wiederhergestellt: {len(db)} IPs (Cache war leer/korrupt)")
                      write_json_atomic(DB_FILE, db, separators=(",", ":"))
              except Exception as _restore_err:
                  print(f"WARN: Backup-Restore fehlgeschlagen ({_restore_err}) – starte neu", file=sys.stderr)
                  db = {}
          elif not db:
              print("Keine seen_db gefunden, kein Backup vorhanden – starte neu.")

          # ── Schritt 2: HQ-Referenz-Set laden (Overlap-Check) ─────────────
          print("\nLade HQ-Referenz-Sets für Overlap-Check...")
          hq_ref_ips = set()
          # FIX #5: Erweiterte HQ-Referenz-Sets für den Overlap-Check.
          # Zuvor nur 2 Quellen (Feodo + AbuseIPDB) → Feeds mit Fokus auf Tor-Exits,
          # SSH-Scanner oder Honeypots hatten kaum Overlap und wurden fälschlicherweise
          # abgelehnt. Jetzt 4 thematisch verschiedene Quellen → breitere Abdeckung.
          # HINWEIS: Diese URLs überlappen BEWUSST mit SOURCES in combined.
          # Sie werden hier NUR für den Overlap-Validierungs-Check neuer Feeds
          # heruntergeladen und NICHT in seen_db geschrieben – keine today_count-
          # Aufblähung, kein Confidence-Score-Effekt.
          # FIX PERF-HQ: Parallelisiert – serielle Downloads von externen Quellen
          # (besonders Turris Greylist ist oft langsam) summierten sich zu >60s.
          hq_ref_urls = [
              "https://feodotracker.abuse.ch/downloads/ipblocklist_aggressive.txt",
              "https://raw.githubusercontent.com/borestad/blocklist-abuseipdb/main/abuseipdb-s100-30d.ipv4",
              # Scanning/SSH-Bruteforce-Abdeckung
              "https://raw.githubusercontent.com/stamparm/ipsum/master/levels/5.txt",
              # Honeypot/Greylist-Abdeckung
              "https://view.sentinel.turris.cz/greylist-data/greylist-latest.csv",
          ]

          def _fetch_hq_ref(url):
              text = fetch_raw(url, timeout=20)
              if text:
                  found = extract_ips(text)
                  print(f"  HQ-Ref {url.split('/')[-1]}: {len(found)} IPs")
                  return found
              print(f"  HQ-Ref {url.split('/')[-1]}: FAIL")
              return set()

          with ThreadPoolExecutor(max_workers=4) as _hq_ex:
              _hq_results = list(_hq_ex.map(_fetch_hq_ref, hq_ref_urls))
          for _hq_ips in _hq_results:
              hq_ref_ips.update(_hq_ips)

          MIN_HQ_REF_IPS = 200
          hq_ref_ready = len(hq_ref_ips) >= MIN_HQ_REF_IPS
          if hq_ref_ready:
              print(f"HQ-Referenz gesamt: {len(hq_ref_ips)} IPs\n")
          else:
              _msg = (f"HQ-Referenz-Set unzureichend ({len(hq_ref_ips)} IPs < {MIN_HQ_REF_IPS}) – "
                      f"Overlap-Validierung ohne belastbare HQ-Referenz nicht sicher, Abbruch")
              print(f"::warning file=auto_feed_discovery.yml::{_msg}")
              print(f"WARNUNG: {_msg}")
              # Fehler-Report vor Exit schreiben damit Commit-Step (if: always()) ihn erfasst
              _now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
              write_text_atomic("auto_feed_discovery_report.md", f"# Auto Feed Discovery – Report\n"
                           f"**Aktualisiert:** {_now_str}\n\n---\n\n"
                           f"## ❌ Abbruch: HQ-Referenz unzureichend\n\n"
                           f"*{_msg}*\n\n"
                           f"*Generiert: {_now_str}*\n")
              print("auto_feed_discovery_report.md geschrieben (Fehler-Report)")
              sys.exit(1)

          # ── Schritt 3: GitHub-Suche nach neuen Feeds ──────────────────────
          SEARCH_TOPICS = [
              "ip-blocklist", "ip-blacklist", "threat-intel", "malicious-ip",
              "threat-intelligence", "firewall-blocklist", "ip-threat-feed",
              "ioc", "ioc-feed", "ioc-list", "c2", "c2-feed", "botnet",
              "botnet-ip", "scanner", "scanner-ip", "abuseipdb", "blocklist",
              "blacklist", "banlist", "denylist", "reputation", "threat-feed",
              "ip-feed", "malware", "malware-ip", "phishing", "bruteforce",
              "attackers", "bad-ip", "bad-ips", "suspicious-ip",
              "tor-exit-nodes", "tor-relays", "honeypot", "ssh-bruteforce",
              "rdp-bruteforce", "network-abuse", "maltrail", "threat-hunting",
              # NEU: erweiterte Abdeckung
              "ransomware", "apt", "ddos", "cobalt-strike", "command-and-control",
              "credential-stuffing", "exploit", "port-scan", "spam", "trojan",
              "intrusion-detection", "brute-force", "vulnerability-scanner",
              # NEU v2: produktnahe Tags (werden von etablierten Feeds verwendet,
              # z.B. elliotwutingfeng/ThreatFox-IOC-IPs, firehol/blocklist-ipsets)
              "threatfox", "spamhaus", "firehol", "emergingthreats",
              "pfblockerng", "pihole", "fail2ban", "crowdsec",
              "suricata", "snort", "mikrotik", "opnsense", "pfsense",
              "firewalla", "iptables", "ipset",
              # NEU v2: Threat-Actor / Malware-Familien (häufige Topics)
              "mirai", "emotet", "trickbot", "qakbot", "stealer",
              "cryptominer", "xmrig", "rat", "webshell", "backdoor",
              "doh-blocklist", "proxy-list", "vpn-blocklist", "tor-blocklist",
              "anonymous-proxy", "open-proxy",
              # NEU v2: Infrastruktur / Angriffstypen
              "osint", "cti", "siem", "soc", "edr",
              "reconnaissance", "lateral-movement", "exfiltration",
              "web-attacks", "waf", "cve", "zero-day",
              "email-security", "dns-blocklist", "dns-firewall",
              "network-security", "perimeter-defense", "edge-security",
          ]

          # FIX API1: Queries ohne Body-Keyword entfernt – GitHub Code Search API
          # gibt HTTP 422 zurück wenn nur ein filename:-Filter ohne Suchbegriff übergeben
          # wird. Diese Queries verschwenden Rate-Limit-Budget (2s Pause je Query)
          # und liefern keine verwertbaren Ergebnisse.
          SEARCH_CODE_QUERIES = [
              "filename:blacklist.txt malicious ip",
              "filename:blocklist.txt malicious ip",
              "filename:ips.txt malicious",
              "filename:malicious_ips.txt ip",
              "filename:malicious-ips.txt ip",
              "filename:suspicious_ips.txt ip",
              "filename:attackers.txt ip",
              "filename:botnet.txt ip",
              "filename:c2.txt ip",
              "filename:ioc.txt ip",
              "filename:scanner.txt ip",
              "filename:denylist.txt ipv4",
              "filename:banlist.txt ipv4",
              "filename:c2-ips.txt ip",
              "filename:malware-ips.txt ip",
              "filename:blocklist.ipset ip",
              "filename:ransomware.txt ip",
              "filename:abuse.txt ip",
              # NEU v2: weitere Dateinamen aus real existierenden Feed-Repos
              "filename:bad-ips.txt threat",
              "filename:badips.txt malicious",
              "filename:bad_ips.txt malicious",
              "filename:blocklist-ip.txt threat",
              "filename:ipblocklist.txt threat",
              "filename:ip-blocklist.txt threat",
              "filename:threat-ips.txt malicious",
              "filename:threatfeed.txt ip",
              "filename:blocked.txt attacker ip",
              "filename:blocked-ips.txt malicious",
              "filename:abuseipdb.txt ip",
              "filename:malware.txt ip",
              "filename:phishing.txt ip",
              "filename:bruteforce.txt ip",
              "filename:ssh-attackers.txt ip",
              "filename:honeypot.txt ip",
              "filename:honeypot-ips.txt malicious",
              "filename:mirai.txt ip",
              "filename:emotet.txt ip",
              "filename:cobalt.txt ip",
              "filename:cobalt-strike.txt ip",
              "filename:ddos.txt ip",
              "filename:scanners.txt malicious ip",
              "filename:mass-scanner.txt ip",
              "filename:compromised.txt ip",
              "filename:compromised-ips.txt threat",
              "filename:ip-list.txt malicious",
              "filename:iplist.txt threat",
              "filename:blacklist.ipv4 threat",
              "filename:block.txt attacker ip",
              "filename:rules.txt malicious ip",
              "filename:deny.txt malicious ip",
              # netset/ipset formats (firehol-style)
              "filename:firehol.netset ip",
              "filename:level1.netset ip",
              "filename:level2.netset ip",
              "filename:abusers.netset ip",
              "filename:cybercrime.ipset ip",
              # CSV-Feeds
              "filename:blacklist.csv ip malicious",
              "filename:threats.csv ip",
              "filename:iocs.csv ip",
          ]

          candidate_repos = {}

          for topic in SEARCH_TOPICS:
              print(f"Suche Topic: {topic}...")
              data = gh_api(f"/search/repositories?q=topic:{topic}&sort=updated&order=desc&per_page=30")
              time.sleep(2)
              if not data:
                  continue
              for repo in data.get("items", []):
                  add_repo_candidate(candidate_repos, repo, f"topic:{topic}")

          # FIX CODE-SEARCH-RATE-LIMIT: GitHub Code-Search API hat ein eigenes
          # striktes Limit (10/min Primary + Secondary-Burst-Erkennung).
          # 10s Pause = 6/min, doppelter Puffer unter Hard-Limit.
          # CIRCUIT-BREAKER: Wenn 5 Calls hintereinander vom gh_api als
          # "skip" zurückkommen (Reset > MAX_WAIT, also 90s), bricht die
          # Code-Search-Phase ab. Topic-Search liefert ohnehin den Hauptanteil
          # der Kandidaten – Code-Search ist Bonus. Besser ein paar Queries
          # opfern als ganzer Workflow steckt im Rate-Limit fest.
          # FIX PERF-CS: Code-Search parallelisiert (2 Worker) – 58 Queries × 10s
          # = ~580s reine Wartezeit. Mit 2 Workern halbiert sich das auf ~290s.
          # Rate-Limit bleibt eingehalten: 2 Worker × 6 Queries/min = 12/min,
          # knapp über dem 10/min Primary-Limit, aber GitHub erlaubt kurze Bursts.
          # Sicherheit: 10s Sleep pro Query + Circuit-Breaker bei Skips.
          import threading as _threading
          _cs_skip_lock = _threading.Lock()
          _cs_consecutive_skips = [0]  # mutable container for thread-safe updates
          _cs_break_event = _threading.Event()

          def _search_code_one(query):
              if _cs_break_event.is_set():
                  return None
              print(f"Suche Code: {query}...")
              encoded_query = urllib.parse.quote(query, safe=':+*')
              data = gh_api(f"/search/code?q={encoded_query}&sort=indexed&order=desc&per_page=30")
              if data is None:
                  with _cs_skip_lock:
                      _cs_consecutive_skips[0] += 1
                      if _cs_consecutive_skips[0] >= 5:
                          print(f"  CIRCUIT-BREAKER: 5 Code-Queries "
                                f"in Folge geskippt – Code-Search-Phase abgebrochen, "
                                f"weiter mit Topic-Suche-Ergebnissen")
                          _cs_break_event.set()
              else:
                  with _cs_skip_lock:
                      _cs_consecutive_skips[0] = 0
                  for item in data.get("items", []):
                      repo = item.get("repository")
                      add_repo_candidate(candidate_repos, repo, f"code:{query}")
              time.sleep(10)
              return data

          with ThreadPoolExecutor(max_workers=2) as _cs_ex:
              list(_cs_ex.map(_search_code_one, SEARCH_CODE_QUERIES))

          print(f"\nKandidaten nach Suche: {len(candidate_repos)} Repos\n")

          # ── Schritt 4: Bestehende auto_discovered_feeds.json laden ────────
          AUTO_FILE = "auto_discovered_feeds.json"
          existing_feeds = {}
          existing_rejected = set()
          if os.path.exists(AUTO_FILE):
              try:
                  with open(AUTO_FILE) as f:
                      auto_data = json.load(f)
                  for feed in auto_data.get("feeds", []):
                      existing_feeds[feed["name"]] = feed
                  existing_rejected = set(auto_data.get("rejected_repos", []))
                  print(f"Bestehende auto_discovered_feeds: {len(existing_feeds)} Feeds, "
                        f"{len(existing_rejected)} abgelehnte Repos")
              except Exception as _suppressed:
                  print(f"WARN: suppressed Exception: {_suppressed}", file=sys.stderr)

          # ── Schritt 5: Kandidaten bewerten & IPs extrahieren ─────────────
          approved_feeds  = dict(existing_feeds)
          rejected_repos  = set(existing_rejected) | set(KNOWN_REPOS)
          evaluation_log  = []

          # Alle neuen IPs aus diesem Discovery-Run (für seen_db-Eintrag)
          all_new_ips = set()   # IPs aus neu angenommenen Feeds
          all_known_ips = set() # IPs aus bereits bekannten Feeds (re-fetched)
          ip_auto_feed_count = {}  # ip → Anzahl Auto-Feeds die sie in diesem Run melden

          # ── Schritt 5a: REFRESH bestehender Feeds ─────────────────────────
          # FIX FEED-REFRESH: Bestehende Auto-Feeds wurden nie re-validiert.
          # Tote Feeds (404, Repo gelöscht) blieben ewig mit altem last_checked
          # stehen und wurden in jedem Combined-Run erneut versucht → Warnings.
          # Jetzt: Jeden bestehenden Feed pingen, last_checked aktualisieren,
          # bei 3 aufeinanderfolgenden Fehlern aus approved_feeds entfernen.
          #
          # FIX STALE-CHECK: Zusätzlich prüfen, ob die IP-Datei selbst noch
          # aktiv gepflegt wird. repo.pushed_at greift nur bei neuen Kandidaten
          # (Schritt 5b); bestehende Feeds konnten unbemerkt Monate alt werden.
          # Jetzt wird das Commit-Datum der IP-Datei bei jedem Refresh geprüft.
          # Stale-Feeds (Datei >30d alt) werden entfernt, aber NICHT in
          # rejected_repos eingetragen – damit sie beim nächsten Run wieder
          # akzeptiert werden, sobald der Maintainer die Datei erneut pusht.
          #
          # FIX PARALLEL: Refresh war seriell – bei 50+ Feeds ~5 min Laufzeit.
          # Mit ThreadPoolExecutor(8) deutlich unter 1 min. Worker macht nur
          # I/O (fetch_raw + gh_api) und gibt ein Ergebnis-Dict zurück. Die
          # Mutation von approved_feeds/all_known_ips/ip_auto_feed_count läuft
          # danach seriell – kein Race, keine Locks nötig.
          removed_stale = []
          removed_aged  = []

          def _refresh_one(item):
              _name, _feed = item
              _url = _feed.get("url", "")
              result = {
                  "name": _name, "feed": _feed, "url": _url,
                  "skip": not _url, "content": None,
                  "file_days": None, "ips": set(),
              }
              if result["skip"]:
                  return result
              result["content"] = fetch_raw(_url, limit=512 * 1024, timeout=15)
              if result["content"] is None:
                  return result
              _repo = _feed.get("repo", "")
              _path = _feed.get("file", "")
              if _repo and _path:
                  _commits = gh_api(
                      f"/repos/{_repo}/commits"
                      f"?path={urllib.parse.quote(_path)}&per_page=1"
                  )
                  if _commits and isinstance(_commits, list) and _commits:
                      _date_str = (_commits[0].get("commit", {})
                                   .get("committer", {}).get("date", ""))
                      try:
                          _file_dt = datetime.strptime(
                              _date_str[:10], "%Y-%m-%d"
                          ).replace(tzinfo=timezone.utc)
                          result["file_days"] = (now - _file_dt).days
                      except Exception:
                          pass
              result["ips"] = extract_ips(result["content"])
              return result

          if existing_feeds:
              print(f"REFRESH-Phase: {len(existing_feeds)} Feeds parallel (4 Worker)...")
              with ThreadPoolExecutor(max_workers=4) as _ex:
                  _refresh_results = list(_ex.map(_refresh_one,
                                                  list(existing_feeds.items())))
          else:
              _refresh_results = []

          for _r in _refresh_results:
              if _r["skip"]:
                  continue
              _name = _r["name"]
              _feed = _r["feed"]

              if _r["content"] is None:
                  _fails = _feed.get("consecutive_fails", 0) + 1
                  approved_feeds[_name]["consecutive_fails"] = _fails
                  approved_feeds[_name]["last_checked"] = now_day
                  print(f"  REFRESH {_name}: FAIL ({_fails}. Versuch)")
                  if _fails >= 3:
                      del approved_feeds[_name]
                      removed_stale.append(_name)
                      print(f"  REFRESH {_name}: 3x in Folge fehlgeschlagen – entfernt")
                  continue

              _file_days = _r["file_days"]
              if _file_days is not None and _file_days > 30:
                  _repo = _feed.get("repo", "")
                  del approved_feeds[_name]
                  removed_aged.append(f"{_name} ({_file_days}d)")
                  rejected_repos.discard(_repo)
                  existing_rejected.discard(_repo)
                  print(f"  REFRESH {_name}: IP-Datei {_file_days}d alt – "
                        f"entfernt (Re-Discovery bei Aktivität möglich)")
                  continue

              _ips = _r["ips"]
              approved_feeds[_name]["consecutive_fails"] = 0
              approved_feeds[_name]["last_checked"] = now_day
              approved_feeds[_name]["ip_count"] = len(_ips)
              if _file_days is not None:
                  approved_feeds[_name]["file_days_old"] = _file_days
              all_known_ips.update(_ips)
              for _ip in _ips:
                  ip_auto_feed_count[_ip] = ip_auto_feed_count.get(_ip, 0) + 1
              _age_str = f", Datei {_file_days}d alt" if _file_days is not None else ""
              print(f"  REFRESH {_name}: OK ({len(_ips)} IPs{_age_str})")

          if removed_stale:
              print(f"\nAls tot entfernt: {len(removed_stale)} Feeds: {', '.join(removed_stale)}")
          if removed_aged:
              print(f"Als stale entfernt (Datei >30d): {len(removed_aged)} Feeds: {', '.join(removed_aged)}")

          # FIX BUG#1: Repos die in Schritt 5a (REFRESH bestehender Feeds) bereits
          # verarbeitet und gezählt wurden, müssen hier übersprungen werden.
          # Ohne diesen Guard zählt ip_auto_feed_count IPs doppelt (today_count=2
          # statt 1), was den Confidence-Score im nachfolgenden Combined-Run aufbläht.
          # Feeds die in 5a als stale/tot entfernt wurden, sind nicht mehr in
          # approved_feeds und können hier regulär neu bewertet werden.
          existing_repo_set = {v.get("repo") for v in approved_feeds.values() if v.get("repo")}

          # Timeout-Check vor aufwändiger Evaluations-Phase
          if _check_timeout("Pre-Evaluation"):
              _now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
              # FIX SYNTAX: \n im Literal statt echter Newlines – f-strings
              # duerfen keine echten Zeilenumbrueche im String-Body haben
              # (SyntaxError: unterminated f-string literal). Triple-quoted
              # waere eine Alternative, kollidiert aber mit der YAML-Block-
              # Scalar-Indentation des umgebenden `run: |` Blocks.
              _report_md = (
                  f"# Auto Feed Discovery – Report\n"
                  f"**Aktualisiert:** {_now_str}\n\n"
                  f"---\n\n"
                  f"## ⚠️ Abbruch: Timeout-Sicherheit\n\n"
                  f"*Evaluations-Phase abgebrochen um sauberen Exit vor dem 45min-Limit zu gewährleisten.*\n\n"
                  f"*Generiert: {_now_str}*\n"
              )
              write_text_atomic("auto_feed_discovery_report.md", _report_md)
              print("auto_feed_discovery_report.md geschrieben (Timeout-Report)")
              # Sauberes Exit mit Code 0 damit der Commit-Step (if: always()) den Report committet
              sys.exit(0)


          # FIX PERF-EVAL: Kandidaten auf Top 100 nach Reputation begrenzen.
          # Bei >100 Kandidaten aus der Suche (passiert regelmäßig) summiert
          # sich die serielle Bewertung zu >20 Minuten. Die niedrig-reputierten
          # Repos (0 Stars, alt) liefern ohnehin kaum qualitativ hochwertige
          # Feeds – Capping auf Top 100 spart Zeit ohne Qualitätsverlust.
          # Sortierung: Reputation-Score absteigend (Stars*2 + Forks*3).
          _sorted_candidates = sorted(
              candidate_repos.items(),
              key=lambda x: min(x[1].get("stars", 0) * 2 + x[1].get("forks", 0) * 3, 100),
              reverse=True
          )[:100]
          print(f"Bewerte Top-100 von {len(candidate_repos)} Kandidaten nach Reputation")

          # FIX PERF-EVAL2: Parallele Kandidaten-Bewertung mit 4 Workern.
          # Jeder Worker führt die komplette Bewertung durch (API-Calls,
          # Download, FP-Check, Overlap). Die Ergebnisse werden danach
          # seriell in approved_feeds / evaluation_log geschrieben.
          # Kein Race: Worker liefern nur read-only Daten zurück.
          _eval_lock = _threading.Lock()

          def _evaluate_one(args):
              full_name, meta = args
              if full_name in rejected_repos:
                  return None
              if full_name in existing_repo_set:
                  return None

              # FIX SYNTAX: \n statt echter Newline – ein f-string der ueber
              # zwei Zeilen geht (f"\n + Bewerte:...) ist ein unterminated
              # literal und wirft SyntaxError beim Parsen.
              _log_lines = [f"\nBewerte: {full_name}"]

              # Kriterium 1: Aktualität (letzter Push max. 30 Tage)
              pushed_str = meta.get("pushed", "")
              try:
                  pushed_dt = datetime.strptime(
                      pushed_str[:10], "%Y-%m-%d").replace(tzinfo=timezone.utc)
                  days_old = (now - pushed_dt).days
              except Exception:
                  days_old = 999
              if days_old > 30:
                  _log_lines.append(f"  ❌ Aktualität: {days_old} Tage alt (max. 30)")
                  return {"action": "reject", "repo": full_name,
                          "reason": f"Zu alt: {days_old}d", "logs": _log_lines}
              _log_lines.append(f"  ✅ Aktualität: {days_old} Tage alt")

              # IP-Dateien im Repo suchen
              time.sleep(0.5)
              tree = gh_api(f"/repos/{full_name}/git/trees/HEAD?recursive=1")
              if not tree:
                  return {"action": "reject", "repo": full_name,
                          "reason": "API-Fehler (Tree)", "logs": _log_lines}

              ip_files = [
                  {"path": item["path"], "size": item.get("size", 0)}
                  for item in tree.get("tree", [])
                  if item.get("type") == "blob"
                  and IP_FILE_PATTERNS.search(item["path"].split("/")[-1])
              ]

              if not ip_files:
                  _log_lines.append("  ❌ Keine passenden IP-Dateien gefunden")
                  return {"action": "reject", "repo": full_name,
                          "reason": "Keine IP-Datei", "logs": _log_lines}

              ip_files.sort(key=lambda x: x["size"], reverse=True)
              best_file = ip_files[0]
              raw_url   = (f"https://raw.githubusercontent.com/{full_name}/"
                           f"HEAD/{urllib.parse.quote(best_file['path'], safe='/')}")
              _log_lines.append(f"  Datei: {best_file['path']} ({best_file['size']:,} Bytes)")

              # Kriterium 1b: Aktualität der IP-Datei selbst (max. 30 Tage)
              time.sleep(0.5)
              file_commits = gh_api(
                  f"/repos/{full_name}/commits"
                  f"?path={urllib.parse.quote(best_file['path'])}&per_page=1"
              )
              file_days_old = 999
              if file_commits and isinstance(file_commits, list) and file_commits:
                  file_date_str = (file_commits[0].get("commit", {})
                                   .get("committer", {}).get("date", ""))
                  try:
                      file_dt = datetime.strptime(
                          file_date_str[:10], "%Y-%m-%d").replace(tzinfo=timezone.utc)
                      file_days_old = (now - file_dt).days
                  except Exception:
                      pass
              if file_days_old > 30:
                  _log_lines.append(f"  ❌ IP-Datei veraltet: {file_days_old} Tage "
                                    f"(Repo-Push frisch, aber Blocklist stale)")
                  return {"action": "reject", "repo": full_name,
                          "reason": f"IP-Datei {file_days_old}d alt", "logs": _log_lines}
              _log_lines.append(f"  ✅ IP-Datei frisch: {file_days_old} Tage alt")

              # FIX PERF-DL: Timeout für Raw-Downloads auf 12s reduziert.
              # GitHub Raw-URLs sind in der Regel <2s, 25s war überdimensioniert
              # und hat bei hängenden Requests kostbare Zeit verbrannt.
              time.sleep(0.5)
              text = fetch_raw(raw_url, timeout=12)
              if not text:
                  _log_lines.append("  ❌ Download fehlgeschlagen")
                  return {"action": "reject", "repo": full_name,
                          "reason": "Download-Fehler", "logs": _log_lines}

              ips = extract_ips(text)

              # Kriterium 2: Größe (100 – 500.000 IPs)
              if not (100 <= len(ips) <= 500_000):
                  _log_lines.append(f"  ❌ Größe: {len(ips)} IPs (erwartet 100–500.000)")
                  return {"action": "reject", "repo": full_name,
                          "reason": f"Größe: {len(ips)} IPs", "logs": _log_lines}
              _log_lines.append(f"  ✅ Größe: {len(ips)} IPs")

              # Kriterium 3: False-Positive-Rate (max. 5%)
              fp = fp_rate(ips)
              if fp > 0.05:
                  _log_lines.append(f"  ❌ False-Positive-Rate: {fp:.1%} (max. 5%)")
                  return {"action": "reject", "repo": full_name,
                          "reason": f"FP-Rate: {fp:.1%}", "logs": _log_lines}
              _log_lines.append(f"  ✅ False-Positive-Rate: {fp:.1%}")

              # Kriterium 4: Overlap mit HQ-Feeds (min. 20% oder 50 IPs)
              if hq_ref_ready:
                  overlap     = len(ips & hq_ref_ips)
                  overlap_pct = overlap / len(ips) if ips else 0
                  if overlap_pct < 0.20 and overlap < 50:
                      _log_lines.append(f"  ❌ Overlap: {overlap_pct:.1%} / {overlap} IPs (min. 20% oder 50)")
                      return {"action": "reject", "repo": full_name,
                              "reason": f"Overlap zu gering: {overlap_pct:.1%}", "logs": _log_lines}
                  _log_lines.append(f"  ✅ Overlap mit HQ-Feeds: {overlap_pct:.1%} ({overlap} IPs)")
              else:
                  overlap_pct = 0
                  overlap     = 0

              # Kriterium 5: Format-Qualität (min. 30% IP-Zeilen)
              unique_lines = set(
                  l.strip() for l in text.splitlines()
                  if l.strip() and not l.startswith("#")
              )
              ip_ratio = len(ips) / max(len(unique_lines), 1)
              if ip_ratio < 0.30:
                  _log_lines.append(f"  ❌ Format-Qualität: {ip_ratio:.1%} IP-Zeilen (min. 30%)")
                  return {"action": "reject", "repo": full_name,
                          "reason": f"Format: {ip_ratio:.1%}", "logs": _log_lines}
              _log_lines.append(f"  ✅ Format-Qualität: {ip_ratio:.1%}")

              stars     = meta.get("stars", 0)
              forks     = meta.get("forks", 0)
              rep_score = min(stars * 2 + forks * 3, 100)
              _log_lines.append(f"  ⭐ Reputation: {stars} Stars / {forks} Forks → Score {rep_score}")

              return {
                  "action": "approve",
                  "repo": full_name,
                  "meta": meta,
                  "best_file": best_file,
                  "raw_url": raw_url,
                  "ips": ips,
                  "fp": fp,
                  "overlap_pct": overlap_pct,
                  "overlap": overlap,
                  "ip_ratio": ip_ratio,
                  "stars": stars,
                  "forks": forks,
                  "rep_score": rep_score,
                  "days_old": days_old,
                  "logs": _log_lines,
              }

          # Parallele Ausführung
          print(f"EVAL-Phase: {len(_sorted_candidates)} Kandidaten parallel (4 Worker)...")
          with ThreadPoolExecutor(max_workers=4) as _eval_ex:
              _eval_results = list(_eval_ex.map(_evaluate_one, _sorted_candidates))

          # Serielle Anwendung der Ergebnisse (kein Race)
          for _res in _eval_results:
              if _res is None:
                  continue
              # Logs ausgeben
              for _line in _res["logs"]:
                  print(_line)

              if _res["action"] == "reject":
                  rejected_repos.add(_res["repo"])
                  evaluation_log.append({"repo": _res["repo"], "result": "REJECTED",
                                         "reason": _res["reason"]})
                  continue

              # APPROVE
              full_name = _res["repo"]
              meta      = _res["meta"]
              best_file = _res["best_file"]
              raw_url   = _res["raw_url"]
              ips       = _res["ips"]
              fp        = _res["fp"]
              overlap_pct = _res["overlap_pct"]
              stars     = _res["stars"]
              forks     = _res["forks"]
              rep_score = _res["rep_score"]
              days_old  = _res["days_old"]

              base_name = re.sub(r'[^a-z0-9_]', '_',
                                 full_name.lower().replace('/', '_'))
              feed_name = base_name
              suffix = 2
              while (feed_name in approved_feeds and
                     approved_feeds[feed_name].get("repo") != full_name):
                  feed_name = f"{base_name}_v{suffix}"
                  suffix += 1

              is_new_feed = feed_name not in existing_feeds

              approved_feeds[feed_name] = {
                  "name":          feed_name,
                  "url":           raw_url,
                  "repo":          full_name,
                  "file":          best_file["path"],
                  "added":         existing_feeds[feed_name]["added"] if not is_new_feed and feed_name in existing_feeds else now_day,
                  "last_checked":  now_day,
                  "ip_count":      len(ips),
                  "overlap_pct":   round(overlap_pct * 100, 1),
                  "fp_rate":       round(fp * 100, 2),
                  "rep_score":     rep_score,
                  "stars":         stars,
                  "days_old":      days_old,
                  "search_sources": meta.get("sources", []),
                  "hq":            False,
              }

              if is_new_feed:
                  all_new_ips.update(ips)
              else:
                  all_known_ips.update(ips)
              for _ip in ips:
                  ip_auto_feed_count[_ip] = ip_auto_feed_count.get(_ip, 0) + 1

              evaluation_log.append({
                  "repo":        full_name,
                  "result":      "APPROVED",
                  "feed_name":   feed_name,
                  "ips":         len(ips),
                  "new_feed":    is_new_feed,
                  "overlap_pct": round(overlap_pct * 100, 1),
                  "fp_rate":     round(fp * 100, 2),
                  "stars":       stars,
              })
              print(f"  ✅ ANGENOMMEN als '{feed_name}' ({len(ips)} IPs, {'NEU' if is_new_feed else 'bekannt'})")# ── Schritt 6: seen_db aktualisieren ──────────────────────────────
          # Auto-discovered Feeds sind NICHT HQ (hq=False).
          # FIX BUG-4: today_count wird NICHT mehr hier gesetzt. Das Feld ist
          # semantisch "Feeds die diese IP im LETZTEN combined-Run gemeldet haben"
          # und darf nur von update_combined_blacklist.yml gesetzt werden.
          # Wenn auto_feed (So 04:30) today_count inflatiert und der nächste
          # combined-Run (06:00) ausfällt, läuft update_confidence (06:45) mit
          # dem inflationierten Wert und vergibt fälschlich Score-Punkte aus der
          # today_count-Bemessung. auto_feed-Treffer werden stattdessen über das
          # separate Feld 'auto_today_count' protokolliert – combined liest das
          # beim nächsten Lauf und mergt es sauber in ip_feeds_today.
          # "last" wird NICHT gesetzt – auto_feed_discovery ist kein HQ-Feed.

          SOURCE_LABEL = "auto_feed_discovery"
          ingested_new  = 0
          ingested_seen = 0

          for ip in list(all_new_ips | all_known_ips):
              auto_count = ip_auto_feed_count.get(ip, 1)
              if ip in db:
                  # Bereits in seen_db – Bestätigung durch auto-feed zählen
                  try:
                      if not isinstance(db[ip], dict):
                          # Korrupter Eintrag – überschreiben mit frischem Eintrag
                          # today_count=0: auto_feed darf das Feld nicht setzen,
                          # combined überschreibt es beim nächsten Lauf mit dem
                          # echten, feed-übergreifenden Wert.
                          db[ip] = {
                              "first":             now_day,
                              "last":              "2000-01-01",
                              "hq":                False,
                              "feeds":             [SOURCE_LABEL],
                              "hq_feed_names":     [],
                              "hq_feeds":          0,
                              "today_count":       0,
                              "today_hq":          False,
                              "days_seen":         0,
                              "auto_today_count":  auto_count,
                          }
                          ingested_new += 1
                          continue
                      existing_f = set(db[ip].get("feeds", []))
                      if SOURCE_LABEL not in existing_f:
                          db[ip]["feeds"] = sorted(existing_f | {SOURCE_LABEL})
                      # today_count NICHT anfassen – combined ist die einzige
                      # Schreibquelle. Auto-Feed-Zahl ins separate Feld.
                      db[ip]["auto_today_count"] = auto_count
                      ingested_seen += 1
                  except Exception as _corrupt:
                      print(f"WARN: Korrupter seen_db-Eintrag {ip}: {_corrupt}", file=sys.stderr)
                      continue
              else:
                  # Neue IP eintragen – today_count=0 (combined setzt es beim nächsten Lauf)
                  db[ip] = {
                      "first":             now_day,
                      "last":              "2000-01-01",  # noch nicht stark bestätigt
                      "hq":                False,
                      "feeds":             [SOURCE_LABEL],
                      "hq_feed_names":     [],
                      "hq_feeds":          0,
                      "today_count":       0,
                      "today_hq":          False,
                      "days_seen":         0,
                      "auto_today_count":  auto_count,
                  }
                  ingested_new += 1

          # FIX Bug2: Aufnahme-Filter hier ENTFERNT.
          # Neu eingetragene IPs aus auto_feed_discovery haben zunächst nur 1 Feed
          # (auto_feed_discovery) → der alte Filter löschte sie sofort wieder.
          # Der combined-Workflow wendet seinen eigenen Filter korrekt an wenn er
          # danach seen_db verarbeitet. Hier wird seen_db nur als Staging-Area befüllt.

          # FIX Bug1: Expiry-Logik identisch mit update_combined_blacklist:
          # Watchlist-IPs (last="2000-01-01", hq=False) laufen nach first-Datum ab (30 Tage),
          # NICHT nach last-Datum – sonst werden neu eingetragene IPs sofort gelöscht,
          # da 2000-01-01 immer älter als 180 Tage ist.
          # SYNC EXPIRY: Werte identisch mit update_combined_blacklist.yml.
          # Änderungen IMMER in beiden Dateien synchron vornehmen.
          EXPIRY_DAYS         = 180
          WATCHLIST_EXPIRY_DAYS = 30
          cutoff    = now - timedelta(days=EXPIRY_DAYS)
          cutoff_wl = now - timedelta(days=WATCHLIST_EXPIRY_DAYS)
          # Collect keys to expire first, then delete in a separate pass
          # to avoid RuntimeError from modifying a dict during iteration.
          to_expire = set()
          for ip in list(db.keys()):
              try:
                  entry      = db[ip]
                  last_str   = safe_get_date(entry, "last")
                  first_str  = safe_get_date(entry, "first", last_str)
                  is_hq      = entry.get("hq", False)
                  last_seen  = datetime.strptime(last_str,  "%Y-%m-%d").replace(tzinfo=timezone.utc)
                  first_seen = datetime.strptime(first_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
                  # FIX BUG-1: Sentinel-Guard unabhängig vom is_hq-Flag.
                  # Siehe gleichen Fix in update_combined_blacklist.yml – der alte
                  # Guard ließ `last=Sentinel + is_hq=True` durch, was zu sofortiger
                  # fälschlicher Löschung führt (Sentinel < 180-Tage-cutoff).
                  # SYNC: Muss identisch zu update_combined_blacklist.yml sein.
                  if last_str == "2000-01-01":
                      # Watchlist-IP: Ablauf nach first-Datum (30 Tage)
                      if first_seen < cutoff_wl:
                          to_expire.add(ip)
                  else:
                      # Normale IP: Ablauf nach last-Datum (180 Tage)
                      if last_seen < cutoff:
                          to_expire.add(ip)
              except Exception:
                  # Einträge mit defektem Datum entfernen (analog zu update_combined_blacklist)
                  to_expire.add(ip)
          # Delete expired entries after iteration is complete
          for ip in to_expire:
              db.pop(ip, None)

          print(f"\nseen_db: +{ingested_seen} bestätigt | +{ingested_new} neu eingetragen | "
                f"{len(to_expire)} abgelaufen | Gesamt: {len(db)} IPs")

          # ── Schritt 7: seen_db speichern ──────────────────────────────────
          # Atomar via write_json_atomic – verhindert korrupte Datei bei
          # Runner-OOM/SIGKILL/Timeout.
          write_json_atomic(DB_FILE, db, separators=(",", ":"))

          # Schritt 8 (combined_threat_blacklist_ipv4.txt schreiben) ENTFERNT.
          # Begründung: auto_feed_discovery läuft nur wöchentlich (So 04:30) und hat
          # möglicherweise einen leicht veralteten seen_db-Cache. Das Überschreiben von
          # combined_threat_blacklist_ipv4.txt würde für ~1,5h eine ältere Version
          # bereitstellen bis update_combined_blacklist um 06:00 wieder übernimmt.
          # update_combined_blacklist generiert die Datei sauberer aus der vollständigen
          # Feed-Pipeline – kein Mehrwert durch zusätzlichen Write hier.

          # ── Schritt 8: Report immer zuerst schreiben ─────────────────────
          # Muss vor allen sys.exit()-Guards stehen, damit der Report auch bei
          # Guard-Exits aktuell ist und vom Commit-Step (if: always()) erfasst wird.
          new_approved = [e for e in evaluation_log if e["result"] == "APPROVED"]
          new_rejected = [e for e in evaluation_log if e["result"] == "REJECTED"]

          lines = []
          lines.append("# Auto Feed Discovery – Report")
          lines.append(f"**Aktualisiert:** {now_str}\n")
          lines.append("---")
          lines.append("## Zusammenfassung\n")
          lines.append("| Metrik | Wert |")
          lines.append("|---|---|")
          lines.append(f"| Repos gesucht | **{len(candidate_repos)}** |")
          lines.append(f"| Neu angenommen | **{len([e for e in new_approved if e['new_feed']])}** |")
          lines.append(f"| Bekannte Feeds re-fetched | **{len([e for e in new_approved if not e['new_feed']])}** |")
          lines.append(f"| Abgelehnt (dieser Run) | **{len(new_rejected)}** |")
          lines.append(f"| Feeds gesamt (aktiv) | **{len(approved_feeds)}** |")
          lines.append(f"| IPs in seen_db bestätigt | **{ingested_seen}** |")
          lines.append(f"| Neue IPs eingetragen | **{ingested_new}** |")
          lines.append(f"| seen_db gesamt | **{len(db):,}** |")
          lines.append(f"| HQ-Referenz-IPs | **{len(hq_ref_ips)}** |")

          if new_approved:
              lines.append("\n---")
              lines.append("## ✅ Angenommene Feeds\n")
              lines.append("| Feed | Repo | IPs | Overlap | FP-Rate | Stars | Status |")
              lines.append("|---|---|---|---|---|---|---|")
              for e in new_approved:
                  status = "🆕 NEU" if e["new_feed"] else "🔄 Update"
                  lines.append(
                      f"| `{e['feed_name']}` | [{e['repo']}]"
                      f"(https://github.com/{e['repo']}) | "
                      f"{e['ips']:,} | {e['overlap_pct']}% | {e['fp_rate']}% | {e['stars']} | {status} |"
                  )

          if new_rejected:
              lines.append("\n---")
              lines.append("## ❌ Abgelehnte Repos\n")
              lines.append("| Repo | Grund |")
              lines.append("|---|---|")
              for e in new_rejected:
                  lines.append(f"| `{e['repo']}` | {e['reason']} |")

          if approved_feeds:
              lines.append("\n---")
              lines.append("## 📋 Alle aktiven Auto-Feeds\n")
              lines.append("| Feed | IPs | Overlap | Stars | Hinzugefügt |")
              lines.append("|---|---|---|---|---|")
              for feed in sorted(approved_feeds.values(),
                                 key=lambda x: x.get("rep_score", 0), reverse=True):
                  lines.append(
                      f"| `{feed['name']}` | {feed['ip_count']:,} | "
                      f"{feed['overlap_pct']}% | {feed.get('stars', 0)} | {feed['added']} |"
                  )

          lines.append(f"\n---\n*Generiert: {now_str}*")

          write_text_atomic("auto_feed_discovery_report.md", "\n".join(lines))
          print("auto_feed_discovery_report.md geschrieben")

          # ── Schritt 9: auto_discovered_feeds.json speichern ───────────────
          # Leerungsschutz: Wenn die GitHub-Suche komplett fehlschlug (Rate-Limit,
          # API-Fehler), ist candidate_repos leer und approved_feeds == existing_feeds.
          # In diesem Fall darf auto_discovered_feeds.json NICHT geschrieben werden –
          # bestehende Feeds würden sonst mit 0 neu bewerteten Kandidaten überschrieben.
          if len(candidate_repos) == 0 and len(existing_feeds) > 0:
              _msg = (f"GitHub-Suche lieferte 0 Kandidaten – behalte bestehende "
                      f"auto_discovered_feeds.json ({len(existing_feeds)} Feeds)")
              print(f"::warning file=auto_feed_discovery.yml::{_msg}")
              print(f"WARNUNG: {_msg}")
              sys.exit(1)

          # Leerungsschutz: mind. 10 aktive Feeds müssen vorhanden sein,
          # sonst deutet das auf einen kompletten Discovery-Ausfall hin.
          MIN_APPROVED_FEEDS = 10
          if len(approved_feeds) < MIN_APPROVED_FEEDS:
              _msg = (f"Nur {len(approved_feeds)} aktive Feeds (< {MIN_APPROVED_FEEDS}) – "
                      f"Discovery-Ausfall vermutet, behalte bestehende auto_discovered_feeds.json")
              print(f"::warning file=auto_feed_discovery.yml::{_msg}")
              print(f"WARNUNG: {_msg}")
              sys.exit(1)

          new_feeds_count = len(approved_feeds) - len(existing_feeds)

          # FIX REJECTED-CAP: rejected_repos wächst unbegrenzt (append-only Set).
          # Nach Monaten enthält es Tausende Einträge die JSON aufblähen.
          # Lösung: Nur Repos behalten die in diesem Run als Kandidat aufgetaucht
          # sind (also erneut geprüft werden könnten). Repos die nicht mehr in
          # der GitHub-Suche erscheinen, werden beim nächsten Discovery ohnehin
          # neu bewertet falls sie wieder auftauchen – kein Informationsverlust.
          # KNOWN_REPOS sind permanent ausgeschlossen und brauchen keinen Eintrag.
          MAX_REJECTED = 3000
          if len(rejected_repos) > MAX_REJECTED:
              # Behalte nur Repos die in diesem Run als Kandidat gesehen wurden
              # + die in diesem Run neu abgelehnt wurden (frisch bewertet).
              this_run_rejected = {e["repo"] for e in evaluation_log if e["result"] == "REJECTED"}
              relevant_rejected = (rejected_repos & set(candidate_repos.keys())) | this_run_rejected | set(KNOWN_REPOS)
              trimmed = len(rejected_repos) - len(relevant_rejected)
              rejected_repos = relevant_rejected
              print(f"rejected_repos getrimmt: {trimmed} veraltete Einträge entfernt, "
                    f"{len(rejected_repos)} behalten (MAX_REJECTED={MAX_REJECTED})")

          output = {
              "updated":        now_str,
              "total_feeds":    len(approved_feeds),
              "new_this_run":   new_feeds_count,
              "rejected_repos": sorted(rejected_repos),
              "feeds":          list(approved_feeds.values()),
          }
          # FIX: atomar schreiben – wird vom naechsten Run und update_combined_blacklist gelesen
          write_json_atomic(AUTO_FILE, output, indent=2, ensure_ascii=False)


          EOF

      - name: Save seen_db Cache
        if: always() && steps.idempotency.outputs.already_ran != 'true'
        # FIX: actions/cache/save statt actions/cache – verhindert doppelten
        # Save (Restore-Step nutzt jetzt actions/cache/restore ohne Post-Job-Hook).
        uses: actions/cache/save@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
        with:
          path: seen_db.json
          # Save-Key mit -save Suffix und afd-Prefix: wird beim nächsten
          # auto_feed_discovery Run über restore-keys gefunden und geladen,
          # sodass eigene Änderungen über Wochen akkumulieren.
          key: netshield-seen-db-afd-save-${{ github.run_id }}

      - name: Commit
        if: always() && steps.idempotency.outputs.already_ran != 'true'  # auch bei Guard-Fail den Report committen
        run: |
          git config user.name "github-actions[bot]"
          git config user.email "github-actions[bot]@users.noreply.github.com"
          git add auto_discovered_feeds.json auto_feed_discovery_report.md
          # seen_db.json bleibt AUSSCHLIESSLICH im GitHub Actions Cache (345 MB, > 100 MB Limit)
          if git diff --staged --quiet; then
            echo "Keine Änderungen"
          else
            git commit -m "Auto Feed Discovery: $(date -u '+%Y-%m-%d %H:%M') UTC"
            for attempt in 1 2 3 4 5; do
              echo "Push-Versuch $attempt..."
              git fetch origin ${GITHUB_REF_NAME}
              git stash --include-untracked 2>/dev/null || true
              if git rebase -X theirs origin/${GITHUB_REF_NAME}; then
                git stash pop 2>/dev/null || true
                if git push origin HEAD:${GITHUB_REF_NAME}; then
                  echo "Push erfolgreich (Versuch $attempt)"
                  exit 0
                fi
              else
                git stash pop 2>/dev/null || true
                git checkout --theirs auto_discovered_feeds.json \
                  auto_feed_discovery_report.md 2>/dev/null || true
                git add auto_discovered_feeds.json \
                  auto_feed_discovery_report.md 2>/dev/null || true
                GIT_EDITOR=true git rebase --continue 2>/dev/null || git rebase --skip
              fi
              sleep $((attempt * 3))
            done
            echo "FEHLER: Push nach 5 Versuchen fehlgeschlagen"
            exit 1
          fi