feat: add CERT-TR (SGB) source via IMAP - Proton Bridge + Gmail (App Password / OAuth2) with mixed RSS

.env.example

@@ -50,3 +50,37 @@ CACHE_PATH=cache/advisories.db
 # Hard caps to bound pagination (avoid silently hitting rate limits without warning)
 # MAX_REPOS=1000
 # MAX_PAGES_PER_REPO=100
+
+# CERT-TR / Siber Güvenlik Başkanlığı via Proton Bridge or Gmail IMAP (never marks as read)
+# When false (default) only GitHub source is active. RSS is mixed (GitHub + CERT-TR) when enabled.
+# Works with both Proton Bridge (127.0.0.1) and Gmail (imap.gmail.com) simultaneously.
+# ENABLE_CERT_TR=false
+# For single account:
+# PROTON_BRIDGE_EMAIL=mail@proton.me
+# PROTON_BRIDGE_PASSWORD=your_bridge_generated_password   # Bridge -> Settings -> Mailbox details (NOT account password)
+# PROTON_IMAP_FOLDER=INBOX                                  # Which folder/label to scan (INBOX, INBOX.CERT-TR, etc.)
+# For multiple accounts (each mail + inbox separately configurable, never marks as read):
+# PROTON_BRIDGE_EMAILS=mail@proton.me,second@proton.me
+# PROTON_BRIDGE_PASSWORDS=bridge_pass1,bridge_pass2         # index-aligned with EMAILS
+# PROTON_IMAP_FOLDERS=INBOX,INBOX.CERT-TR                    # index-aligned; missing -> defaults to PROTON_IMAP_FOLDER
+# Bridge connection (usually don't change):
+# PROTON_BRIDGE_HOST=127.0.0.1
+# PROTON_IMAP_PORT=1143
+# PROTON_IMAP_SECURITY=STARTTLS   # STARTTLS | SSL | NONE (Bridge uses STARTTLS, self-signed)
+# Gmail alternative (same CERT-TR mails via Gmail IMAP, also never marks as read)
+# GMAIL_EMAIL=your@gmail.com
+# GMAIL_APP_PASSWORD=abcd efgh ijkl mnop   # Gmail App Password (16 chars, spaces preserved; for multi use comma)
+# GMAIL_IMAP_FOLDER=INBOX
+# For multiple Gmail accounts (each mail + inbox separately):
+# GMAIL_EMAILS=your@gmail.com,second@gmail.com
+# GMAIL_APP_PASSWORDS=app_pass1,"app pass 2 with spaces"   # comma-separated, spaces inside preserved
+# GMAIL_IMAP_FOLDERS=INBOX,INBOX
+# Gmail IMAP (usually don't change):
+# GMAIL_IMAP_HOST=imap.gmail.com
+# GMAIL_IMAP_PORT=993
+# GMAIL_IMAP_SECURITY=SSL   # SSL for Gmail, STARTTLS for Proton Bridge
+
+# Filtering & limits (applies to both Proton and Gmail):
+# CERT_TR_SENDER_ALLOWLIST=siberguvenlik.gov.tr
+# CERT_TR_MAX_MAILS=200
+# CERT_TR_SEARCH_DAYS=   # empty = all time, or e.g. 30 for last 30 days

pyproject.toml

@@ -14,7 +14,8 @@ dependencies = [
     "pydantic-settings>=2.4",
     "python-dotenv>=1.0",
     "click>=8.0",
-
+    "google-auth>=2.30",
+    "google-auth-oauthlib>=1.1",
 ]
 
 [project.optional-dependencies]

release.sh

@@ -22,7 +22,7 @@ echo "Tag oluşturuluyor..."
 git tag -f "${TAG}"
 
 echo "Push ediliyor..."
-git push origin master "${TAG}"
+git push origin main "${TAG}"
 
 echo "GitHub Release oluşturuluyor..."
 gh release delete "${TAG}" --yes --cleanup-tag 2>/dev/null || true

scripts/fixtures_generator.py

@@ -308,7 +308,7 @@ async def main_async(args: argparse.Namespace) -> int:
                 p.unlink(missing_ok=True)
         return 0
     except (httpx.HTTPError, OSError, ValueError, RuntimeError) as e:
-        logger.error("fixtures_generator failed: %s", _redact(str(e)), exc_info=True)
+        logger.exception("fixtures_generator failed")
         print(f"ERROR: {_redact(str(e))}", file=sys.stderr)
         import traceback
 

src/advisory_rss/auth/gmail.py

@@ -0,0 +1,258 @@
+#Gmail OAuth helpers
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from advisory_rss.config.constants import (
+    DEFAULT_GMAIL_OAUTH_CACHE,
+    GMAIL_OAUTH_SCOPES,
+    GMAIL_OAUTH_TOKEN_URI,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _load_oauth_file(path: str | Path) -> dict[str, Any]:
+    p = Path(path)
+    if not p.exists():
+        return {}
+    try:
+        data = json.loads(p.read_text(encoding="utf-8"))
+        return data if isinstance(data, dict) else {}
+    except (OSError, ValueError, json.JSONDecodeError) as e:
+        logger.debug("Failed to load Gmail OAuth file %s: %s", path, e)
+        return {}
+
+
+def _save_oauth_file(path: str | Path, data: dict[str, Any]) -> None:
+    p = Path(path)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    # Restrict permissions (600) for token file
+    try:
+        p.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+        try:
+            p.chmod(0o600)
+        except OSError as e:
+            logger.debug("chmod failed for %s: %s", p, e)
+    except OSError as e:
+        logger.warning("Failed to save Gmail OAuth file %s: %s", p, e)
+        raise
+
+
+def save_gmail_oauth_token(
+    token_file: str | Path,
+    email: str,
+    refresh_token: str,
+    client_id: str,
+    client_secret: str,
+    access_token: str | None = None,
+    expiry: str | None = None,
+) -> None:
+    data = _load_oauth_file(token_file)
+    # Preserve top-level client_id/secret if not per-account
+    if client_id:
+        data["client_id"] = client_id
+    if client_secret:
+        data["client_secret"] = client_secret
+    if "accounts" not in data or not isinstance(data["accounts"], dict):
+        data["accounts"] = {}
+    acct = data["accounts"].get(email, {})
+    if not isinstance(acct, dict):
+        acct = {}
+    acct["refresh_token"] = refresh_token
+    if access_token:
+        acct["access_token"] = access_token
+    if expiry:
+        acct["expiry"] = expiry
+    data["accounts"][email] = acct
+    _save_oauth_file(token_file, data)
+    logger.info("Saved Gmail OAuth token for %s to %s", email, token_file)
+
+
+def get_refresh_token_for_email(
+    email: str,
+    settings: Any,
+) -> tuple[str | None, str | None, str | None]:
+    #Return (refresh_token, client_id, client_secret) for email.
+  #Checks env vars first (GMAIL_OAUTH_REFRESH_TOKEN etc), then file.
+    #For multi-account, checks GMAIL_OAUTH_REFRESH_TOKENS aligned with GMAIL_EMAILS.
+    # Check env single/multi
+    # settings may have gmail_oauth_refresh_token(s) and client_id/secret
+    try:
+        file_data = _load_oauth_file(
+            getattr(settings, "gmail_oauth_token_file", DEFAULT_GMAIL_OAUTH_CACHE)
+        )
+        # Try file per-account
+        accounts = file_data.get("accounts") if isinstance(file_data, dict) else None
+        if isinstance(accounts, dict) and email in accounts:
+            info = accounts[email]
+            if isinstance(info, dict) and info.get("refresh_token"):
+                cid = file_data.get("client_id") or getattr(settings, "gmail_oauth_client_id", None)
+                csec = file_data.get("client_secret") or getattr(
+                    settings, "gmail_oauth_client_secret", None
+                )
+                return (
+                    str(info["refresh_token"]),
+                    str(cid or "") if cid else None,
+                    str(csec or "") if csec else None,
+                )
+            if isinstance(info, str):
+                cid = file_data.get("client_id") or getattr(settings, "gmail_oauth_client_id", None)
+                csec = file_data.get("client_secret") or getattr(
+                    settings, "gmail_oauth_client_secret", None
+                )
+                return info, str(cid or "") if cid else None, str(csec or "") if csec else None
+        # Check flat file
+        if isinstance(file_data, dict) and email in file_data and isinstance(file_data[email], str):
+            cid = file_data.get("client_id") or getattr(settings, "gmail_oauth_client_id", None)
+            csec = file_data.get("client_secret") or getattr(
+                settings, "gmail_oauth_client_secret", None
+            )
+            return (
+                str(file_data[email]),
+                str(cid or "") if cid else None,
+                str(csec or "") if csec else None,
+            )
+
+        # Check env multi
+        gmail_emails = getattr(settings, "gmail_emails", None)
+        gmail_refresh_tokens = getattr(settings, "gmail_oauth_refresh_tokens", None)
+        if gmail_emails and gmail_refresh_tokens:
+            # _split_list is private, but we can emulate
+            import re
+
+            emails = [e.strip() for e in re.split(r"[,\s;]+", gmail_emails) if e.strip()]
+            rts = [r.strip() for r in gmail_refresh_tokens.split(",") if r.strip()]
+            try:
+                idx = [e.lower() for e in emails].index(email.lower())
+                if idx < len(rts):
+                    cid = getattr(settings, "gmail_oauth_client_id", None)
+                    csec = getattr(settings, "gmail_oauth_client_secret", None)
+                    return (
+                        rts[idx],
+                        str(cid or "") if cid else None,
+                        str(csec or "") if csec else None,
+                    )
+            except ValueError:
+                pass
+
+        # Check env single
+        single_rt = getattr(settings, "gmail_oauth_refresh_token", None)
+        if single_rt and single_rt.strip():
+            gmail_email = getattr(settings, "gmail_email", None)
+            if gmail_email and gmail_email.strip().lower() == email.lower():
+                cid = getattr(settings, "gmail_oauth_client_id", None)
+                csec = getattr(settings, "gmail_oauth_client_secret", None)
+                return (
+                    single_rt.strip(),
+                    str(cid or "") if cid else None,
+                    str(csec or "") if csec else None,
+                )
+            cid = getattr(settings, "gmail_oauth_client_id", None)
+            csec = getattr(settings, "gmail_oauth_client_secret", None)
+            if not gmail_emails:
+                return (
+                    single_rt.strip(),
+                    str(cid or "") if cid else None,
+                    str(csec or "") if csec else None,
+                )
+    except (OSError, ValueError, AttributeError) as e:
+        logger.debug("get_refresh_token failed for %s: %s", email, e)
+    return None, None, None
+
+
+def fetch_access_token(
+    refresh_token: str,
+    client_id: str,
+    client_secret: str,
+    token_uri: str = GMAIL_OAUTH_TOKEN_URI,
+) -> tuple[str, int | None]:
+    #Exchange refresh token for access token via Google OAuth2.
+    if not refresh_token or not client_id or not client_secret:
+        raise ValueError("Missing OAuth credentials (refresh_token/client_id/secret)")
+
+    data = {
+        "client_id": client_id,
+        "client_secret": client_secret,
+        "refresh_token": refresh_token,
+        "grant_type": "refresh_token",
+    }
+    try:
+        resp = httpx.post(token_uri, data=data, timeout=15.0)
+        resp.raise_for_status()
+        j = resp.json()
+        access_token = j.get("access_token")
+        expires_in = j.get("expires_in")
+        if not access_token:
+            raise ValueError(f"No access_token in response: {j}")
+        return str(access_token), int(expires_in) if expires_in else None
+    except httpx.HTTPError as e:
+        logger.warning("Failed to fetch Gmail access token: %s", e)
+        raise
+    except (ValueError, TypeError) as e:
+        logger.warning("Invalid token response: %s", e)
+        raise
+
+
+def build_xoauth2_string(email: str, access_token: str) -> str:
+    """Build base64-encoded XOAUTH2 string for IMAP AUTHENTICATE."""
+    # Format: user=<email>\x01auth=Bearer <token>\x01\x01
+    raw = f"user={email}\x01auth=Bearer {access_token}\x01\x01"
+    return base64.b64encode(raw.encode("utf-8")).decode("utf-8")
+
+
+def run_gmail_oauth_flow(
+    client_id: str,
+    client_secret: str,
+    email: str | None = None,
+) -> tuple[str, str, str | None]:
+    #Run InstalledAppFlow to obtain refresh token.
+
+    #Returns (refresh_token, access_token, expiry).
+    #Requires google-auth-oauthlib to be installed.
+    #Opens browser for user consent.
+
+    try:
+        from google_auth_oauthlib.flow import InstalledAppFlow
+    except ImportError as e:
+        raise ImportError(
+            "google-auth-oauthlib is required for Gmail OAuth. Install with: pip install google-auth google-auth-oauthlib"
+        ) from e
+
+    # Use the same scopes as constants
+    client_config = {
+        "installed": {
+            "client_id": client_id,
+            "client_secret": client_secret,
+            "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+            "token_uri": GMAIL_OAUTH_TOKEN_URI,
+            "redirect_uris": ["http://localhost", "urn:ietf:wg:oauth:2.0:oob"],
+        }
+    }
+
+    flow = InstalledAppFlow.from_client_config(client_config, scopes=GMAIL_OAUTH_SCOPES)
+    creds = flow.run_local_server(
+        port=0,
+        prompt="consent",
+        access_type="offline",
+        include_granted_scopes="true",
+    )
+    refresh_token = getattr(creds, "refresh_token", None)
+    access_token = getattr(creds, "token", None)
+    expiry = getattr(creds, "expiry", None)
+    expiry_str = expiry.isoformat() if expiry else None
+    if not refresh_token:
+        logger.warning(
+            "OAuth flow did not return a refresh_token (maybe already consented). Try revoking access and retrying."
+        )
+        raise ValueError(
+            "No refresh_token returned from OAuth flow. Ensure you revoke previous consent in Google Account and retry with prompt=consent."
+        )
+    return str(refresh_token), str(access_token or ""), expiry_str

src/advisory_rss/cache/models.py

@@ -1,5 +1,3 @@
-"""Normalized advisory dataclass and cache meta."""
-
 from __future__ import annotations
 
 from dataclasses import asdict, dataclass, field
@@ -32,6 +30,9 @@ class NormalizedAdvisory:
     publisher_login: str | None = None
     identifiers: list[dict[str, str]] = field(default_factory=list)
     raw: dict[str, Any] = field(default_factory=dict, repr=False)
+    # Multi-source extension
+    source: str = "github"
+    extra_cves: list[str] = field(default_factory=list)
 
     def sort_key(self) -> datetime:
         # Use updated_at > published_at > created_at > distant past
@@ -92,6 +93,8 @@ def _parse(v: str | None) -> datetime | None:
             publisher_login=data.get("publisher_login"),
             identifiers=list(data.get("identifiers") or []),
             raw=data.get("raw") or {},
+            source=data.get("source") or "github",
+            extra_cves=list(data.get("extra_cves") or []),
         )
         return adv