🐍 보안 권고문 모니터링 자동화 — 의존성 0으로 만드는 RSS 수집기

# ❌ 취약한 코드 — 일시 장애 한 번에 "신규 0건"으로 조용히 실패
def fetch(url):
    return urllib.request.urlopen(url, timeout=20).read().decode()

# ✅ 안전한 코드 — 오류를 분류해서 재시도할 것만 재시도
HTTP_TIMEOUT, HTTP_RETRIES, HTTP_BACKOFF = 20, 3, 5

def fetch(url: str) -> str | None:
    last = None
    for attempt in range(1, HTTP_RETRIES + 1):
        try:
            req = urllib.request.Request(url, headers={"User-Agent": UA})
            with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as res:
                return res.read().decode("utf-8", errors="replace")
        except urllib.error.HTTPError as e:
            if e.code == 404:
                return None            # 피드 이동 — 호출부에서 표면화
            if not (500 <= e.code < 600):
                raise                  # 4xx는 즉시 드러낸다
            last = e
        except (urllib.error.URLError, TimeoutError, OSError) as e:
            last = e                   # 네트워크성 → 재시도
        if attempt < HTTP_RETRIES:
            time.sleep(HTTP_BACKOFF * attempt)   # 5s, 10s 선형 백오프
    raise last

import xml.etree.ElementTree as ET

def parse_rss(xml_text: str) -> list[dict]:
    channel = ET.fromstring(xml_text).find("channel")
    if channel is None:
        return []
    entries = []
    for item in channel.findall("item"):
        entries.append({
            "title": (item.findtext("title") or "").strip(),
            "link": (item.findtext("link") or "").strip(),
            "pubDate": (item.findtext("pubDate") or "").strip(),
            "description": (item.findtext("description") or "").strip(),
        })
    return entries

SOURCES = {
    "KISA": {
        "rss": "https://www.boho.or.kr/kr/rss.do?bbsId=B0000133",
        # 게시글 링크의 nttId 쿼리 파라미터가 고유 ID
        "entry_id": lambda item: _query_param(item["link"], "nttId"),
    },
    "FORTINET": {
        "rss": "https://filestore.fortinet.com/fortiguard/rss/ir.xml",
        # 링크 끝 FG-IR-YY-NNN이 고유 ID
        "entry_id": lambda item: item["link"].rstrip("/").rsplit("/", 1)[-1],
    },
}

CREATE TABLE IF NOT EXISTS advisories (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    source      TEXT NOT NULL,
    entry_id    TEXT NOT NULL,
    title       TEXT NOT NULL,
    url         TEXT NOT NULL,
    published   TEXT,
    cves        TEXT,
    detected_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
    notified    INTEGER NOT NULL DEFAULT 0,
    UNIQUE(source, entry_id)          -- 이 한 줄이 "이미 봤는지" 판단 전부
);

for entry in parse_rss(xml_text):
    entry_id = src["entry_id"](entry)
    try:
        conn.execute(
            "INSERT INTO advisories (source, entry_id, title, url, published, cves)"
            " VALUES (?,?,?,?,?,?)",
            (key, entry_id, entry["title"], entry["link"],
             entry["pubDate"], extract_cves(entry)))
        new_count += 1
    except sqlite3.IntegrityError:
        pass  # 이미 본 항목 — 조용히 스킵

CVE_RE = re.compile(r"CVE-\d{4}-\d{4,7}")

def extract_cves(entry: dict) -> str:
    found = CVE_RE.findall(entry["title"] + " " + entry["description"])
    return ",".join(sorted(set(found)))

first_run = conn.execute("SELECT COUNT(*) FROM advisories").fetchone()[0] == 0
new_rows = scan(conn)
if first_run:
    mark_notified(conn, new_rows)
    print(f"첫 실행 seed 완료 — {len(new_rows)}건 적재 (알림 생략)")
    return

# Linux cron — 매일 09:00
0 9 * * * cd /path/to/watcher && python watcher.py >> data/cron.log 2>&1

# Windows 작업 스케줄러
schtasks /Create /TN advisory-watcher /SC DAILY /ST 09:00 `
  /TR "python C:\path\to\watcher.py"