with newlines before stripping
+ if not html_body:
+ return ""
+ # normalize quoted-printable already decoded to string
+ s = html_body
+ # Replace block tags with newline
+ s = re.sub(r"(?i)<\s*br\s*/?\s*>", "\n", s)
+ s = re.sub(r"(?i)", "\n\n", s)
+ s = re.sub(r"(?i)", "\n", s)
+ # Remove all tags
+ s = re.sub(r"<[^>]+>", " ", s)
+ # Decode html entities
+ s = html.unescape(s)
+ # Collapse whitespace but keep single newlines
+ s = s.replace("\r", "")
+ # Collapse multiple spaces
+ s = re.sub(r"[ \t]+", " ", s)
+ s = re.sub(r"\n{3,}", "\n\n", s)
+ s = re.sub(r" *\n *", "\n", s)
+ return s.strip()
+
+def _extract_body(msg: EmailMessage) -> tuple[str, str | None]:
+ # Walk parts; prefer html then plain; handle multipart
+ plain: str | None = None
+ html_raw: str | None = None
+
+ if msg.is_multipart():
+ for part in msg.walk():
+ ctype = part.get_content_type()
+ # skip containers
+ if ctype in ("multipart/alternative", "multipart/mixed", "multipart/related"):
+ continue
+ disp = (part.get_content_disposition() or "").lower()
+ if disp == "attachment":
+ continue
+ payload = part.get_payload(decode=True) # type: ignore[assignment]
+ if payload is None:
+ continue
+ charset = part.get_content_charset() or "utf-8"
+ # try charset fallback chain
+ text = None
+ assert isinstance(payload, (bytes, bytearray))
+ for enc in (charset, "utf-8", "windows-1254", "iso-8859-9", "iso-8859-1"):
+ try:
+ text = payload.decode(enc, errors="replace") # type: ignore[union-attr]
+ break
+ except LookupError:
+ continue
+ except (ValueError, UnicodeDecodeError):
+ continue
+ if text is None:
+ text = payload.decode("utf-8", errors="replace") # type: ignore[union-attr]
+ if ctype == "text/html" and html_raw is None:
+ html_raw = text
+ elif ctype == "text/plain" and plain is None:
+ plain = text
+ # fallback: if html but no plain, convert html to plain
+ if html_raw and not plain:
+ plain = _html_to_text(html_raw)
+ if plain and not html_raw:
+ # keep html_raw None
+ pass
+ return plain or "", html_raw
+ else:
+ ctype = msg.get_content_type()
+ payload = msg.get_payload(decode=True) # type: ignore[assignment]
+ if payload is None:
+ # get_payload may return str
+ payload_str = msg.get_payload()
+ if isinstance(payload_str, str):
+ # quoted-printable already?
+ return payload_str, None
+ return "", None
+ charset = msg.get_content_charset() or "utf-8"
+ text = None
+ assert isinstance(payload, (bytes, bytearray))
+ for enc in (charset, "utf-8", "windows-1254", "iso-8859-9"):
+ try:
+ text = payload.decode(enc, errors="replace") # type: ignore[union-attr]
+ break
+ except LookupError:
+ continue
+ if text is None:
+ text = payload.decode("utf-8", errors="replace") # type: ignore[union-attr]
+ if ctype == "text/html":
+ return _html_to_text(text), text
+ return text, None
+
+def _extract_cves(text: str) -> list[str]:
+ found = CVE_RE.findall(text or "")
+ # normalize upper
+ norm = [c.upper() for c in found]
+ # dedup preserve order
+ seen = set()
+ out = []
+ for c in norm:
+ if c not in seen:
+ seen.add(c)
+ out.append(c)
+ return out
+
+def _extract_cwe(text: str) -> str | None:
+ m = CWE_RE.search(text or "")
+ return m.group(0).upper() if m else None
+
+def _extract_product(text: str) -> str | None:
+ if not text:
+ return None
+ m = PRODUCT_RE.search(text)
+ if m:
+ prod = m.group(1).strip()
+ # cleanup quotes artifacts
+ prod = prod.strip("“”\"' ")
+ if prod and len(prod) > 1:
+ return prod
+ # fallback: find quoted strings containing dash or lowercase + not CWE/CVE
+ quoted = QUOTED_RE.findall(text)
+ candidates = []
+ for q in quoted:
+ q = q.strip()
+ if not q or q.upper().startswith("CWE") or q.upper().startswith("CVE"):
+ continue
+ if "TÜBİTAK" in q or "Siber" in q or len(q) > 40:
+ continue
+ # product usually contains - or _
+ if "-" in q or "_" in q or q.islower() or "/" in q:
+ candidates.append(q)
+ if candidates:
+ # shortest that looks like package name
+ candidates.sort(key=len)
+ return candidates[0].strip()
+ return None
+
+def _build_summary(
+ subject_decoded: str, product: str | None, cve: str | None, cwe: str | None
+) -> str:
+ # If product + CVE, build canonical summary, else decoded subject
+ if product and cve:
+ base = f"{product}: {cve}"
+ if cwe:
+ base += f" ({cwe})"
+ return base
+ if cve and cwe:
+ return f"{cve} - {cwe}"
+ if cve:
+ return f"Zafiyet Bildirimi {cve}"
+ # fallback to subject trimmed
+ s = (subject_decoded or "").strip()
+ if s:
+ # Remove excessive prefix "Zafiyet Bildiriminiz Hakkında"
+ s = re.sub(r"^\s*Zafiyet Bildiriminiz Hakkında\s*[:\-]?\s*", "", s, flags=re.IGNORECASE)
+ s = s.strip(" ()")
+ if s:
+ return s
+ return subject_decoded.strip() or "CERT-TR Zafiyet Bildirimi"
+
+def parse_cert_tr_email(raw_bytes: bytes) -> NormalizedAdvisory | None:
+ """Parse raw RFC822 bytes from Proton/Bridge or Gmail into NormalizedAdvisory.
+
+ Returns None if no CVE is assigned (strict filter to avoid false positives).
+ Uses CVE-based id for dedup (CERT-TR-CVE-...).
+ Never marks mail read.
+ """
+ try:
+ msg = BytesParser(policy=policy.default).parsebytes(raw_bytes)
+ except (ValueError, TypeError, AttributeError) as e:
+ logger.warning("Failed to parse email bytes: %s", e)
+ return None
+
+ # Headers
+ subject_raw = msg.get("Subject") or ""
+ subject = decode_rfc2047(str(subject_raw))
+ from_raw = msg.get("From") or ""
+ from_decoded = decode_rfc2047(str(from_raw))
+ reply_to = decode_rfc2047(str(msg.get("Reply-To") or ""))
+ message_id = (msg.get("Message-Id") or msg.get("Message-ID") or "").strip()
+ # also X-Pm-External-Id fallback
+ if not message_id:
+ message_id = (msg.get("X-Pm-External-Id") or "").strip()
+ date_hdr = msg.get("Date") or msg.get("X-Pm-Date") or ""
+ date_dt = _parse_date(str(date_hdr))
+
+ # Body
+ plain_text, html_raw = _extract_body(msg)
+ # Combine subject + plain for CVE search (subject often contains CVE)
+ combined_text = f"{subject}\n{plain_text}"
+ if html_raw:
+ combined_text += f"\n{html_raw}"
+
+ cves = _extract_cves(combined_text)
+ cve_primary = cves[0] if cves else None
+ cwe = _extract_cwe(combined_text)
+ product = _extract_product(combined_text)
+
+ if not cve_primary and message_id:
+ m = CVE_RE.search(message_id)
+ if m:
+ cve_primary = m.group(0).upper()
+ if cve_primary not in cves:
+ cves.append(cve_primary)
+
+ if not cve_primary:
+ logger.debug(
+ "Skipping CERT-TR mail without CVE subject=%r from=%r",
+ subject[:60],
+ from_decoded[:60],
+ )
+ return None
+
+ ghsa_id = f"CERT-TR-{cve_primary.upper()}"
+
+ # Extra CVEs beyond primary
+ extra_cves = [c for c in cves if c != cve_primary]
+
+ # Summary & description
+ summary = _build_summary(subject, product, cve_primary, cwe)
+ # Description: plain_text truncated 8000 chars for RSS, keep full
+ description = plain_text.strip()
+ if not description and html_raw:
+ description = _html_to_text(html_raw)
+ if not description:
+ description = combined_text[:2000]
+
+ # References
+ refs: list[str] = []
+ if cve_primary:
+ refs.append(f"https://nvd.nist.gov/vuln/detail/{cve_primary}")
+ refs.append(f"https://cve.mitre.org/cgi-bin/cvename.cgi?name={cve_primary}")
+ # also include siberguvenlik portal (generic)
+ refs.append("https://siberguvenlik.gov.tr")
+
+ # Links: CVE-based NVD link
+ html_url = f"https://nvd.nist.gov/vuln/detail/{cve_primary}"
+
+ # Severity: try to infer from text (kritik/yüksek/orta/düşük)
+ severity = None
+ low_text = (plain_text or "").lower()
+ if "kritik" in low_text or "critical" in low_text:
+ severity = "critical"
+ elif "yüksek" in low_text or "high" in low_text:
+ severity = "high"
+ elif "orta" in low_text or "medium" in low_text or "moderate" in low_text:
+ severity = "medium"
+ elif "düşük" in low_text or "low" in low_text:
+ severity = "low"
+
+ # Package extraction: ecosystem maybe?
+ # infer ecosystem from product name: pardus => deb? keep None
+ package_ecosystem = None
+ package_name = product
+
+ # State: CERT-TR mails are allocations; consider "published" but note awaiting patch
+ state = "published"
+
+ # Updated/published = date
+ published_at = date_dt
+ updated_at = date_dt
+ created_at = date_dt
+
+ # Author/publisher
+ author_login = from_decoded or "cve@siberguvenlik.gov.tr"
+ # Try to extract email addr only
+ m = re.search(r"<([^>]+@[^>]+)>", author_login)
+ if m:
+ author_login = m.group(1).strip()
+ publisher_login = "siberguvenlik.gov.tr"
+
+ # identifiers list
+ identifiers = []
+ for c in cves:
+ identifiers.append({"type": "CVE", "value": c})
+ if cwe:
+ identifiers.append({"type": "CWE", "value": cwe})
+
+ # raw preservation (redacted password not in raw)
+ raw: dict[str, Any] = {
+ "message_id": message_id,
+ "subject": subject,
+ "from": from_decoded,
+ "reply_to": reply_to,
+ "date": str(date_hdr),
+ "cves": cves,
+ "cwe": cwe,
+ "product": product,
+ "html_raw_snippet": (html_raw[:2000] if html_raw else "")[:2000],
+ "plain_snippet": plain_text[:2000],
+ "x_pm_external_id": str(msg.get("X-Pm-External-Id") or ""),
+ "x_pm_internal_id": str(msg.get("X-Pm-Internal-Id") or ""),
+ }
+
+ adv = NormalizedAdvisory(
+ ghsa_id=ghsa_id,
+ cve_id=cve_primary,
+ summary=summary,
+ description=description,
+ severity=severity,
+ package_ecosystem=package_ecosystem,
+ package_name=package_name,
+ vulnerable_version_range=None,
+ patched_versions=None,
+ affected_versions=None,
+ published_at=published_at,
+ updated_at=updated_at,
+ withdrawn_at=None,
+ created_at=created_at,
+ state=state,
+ html_url=html_url,
+ repository_url=None,
+ repository_full_name=f"cert-tr/{product}" if product else "cert-tr/siberguvenlik",
+ references=refs,
+ author_login=author_login,
+ publisher_login=publisher_login,
+ identifiers=identifiers,
+ raw=raw,
+ source="cert-tr",
+ extra_cves=extra_cves,
+ )
+ logger.debug("Parsed CERT-TR mail ghsa_id=%s cve=%s product=%r", ghsa_id, cve_primary, product)
+ return adv
+
+
+def is_cert_tr_sender(from_header: str, allowlist: list[str]) -> bool:
+ if not from_header:
+ return False
+ decoded = decode_rfc2047(from_header)
+ domain = _extract_domain(decoded)
+ if not domain:
+ # fallback raw lower
+ domain = from_header.lower()
+ for allowed in allowlist:
+ allowed = allowed.lower().strip()
+ if not allowed:
+ continue
+ if domain == allowed or domain.endswith("." + allowed) or allowed in domain:
+ return True
+ # also substring check for safety
+ if allowed in from_header.lower():
+ return True
+ return False
diff --git a/src/advisory_rss/cert_tr/source.py b/src/advisory_rss/cert_tr/source.py
new file mode 100644
index 0000000..d0faee4
--- /dev/null
+++ b/src/advisory_rss/cert_tr/source.py
@@ -0,0 +1,107 @@
+#CERT-TR source orchestrator: IMAP fetch -> parser -> NormalizedAdvisory list.
+
+from __future__ import annotations
+
+import imaplib
+import logging
+from typing import Any
+
+from advisory_rss.cache.models import NormalizedAdvisory
+from advisory_rss.cert_tr import imap as cert_imap
+from advisory_rss.cert_tr.parser import is_cert_tr_sender, parse_cert_tr_email
+from advisory_rss.config.settings import Settings
+
+logger = logging.getLogger(__name__)
+
+
+class CertTrSource:
+ def __init__(self, settings: Settings):
+ self.settings = settings
+
+ def fetch_all(self) -> tuple[list[NormalizedAdvisory], dict[str, Any]]:
+ #Fetch from all configured Proton accounts.
+ diag: dict[str, Any] = {
+ "accounts": 0,
+ "raw_fetched": 0,
+ "parsed": 0,
+ "filtered_sender": 0,
+ "errors": [],
+ }
+ if not self.settings.enable_cert_tr:
+ logger.debug("CERT-TR disabled (ENABLE_CERT_TR=false)")
+ return [], diag
+
+ accounts = self.settings.get_cert_tr_accounts()
+ if not accounts:
+ msg = "ENABLE_CERT_TR=true but no PROTON_BRIDGE_EMAIL/EMAILS or GMAIL_EMAIL/EMAILS configured"
+ logger.warning(msg)
+ diag["errors"].append(msg)
+ return [], diag
+
+ allowlist = self.settings.cert_tr_sender_list()
+ max_mails = self.settings.cert_tr_max_mails
+ search_days = self.settings.cert_tr_search_days
+
+ all_advs: list[NormalizedAdvisory] = []
+ per_account_max = max_mails
+ if len(accounts) > 1:
+ per_account_max = max(20, max_mails // len(accounts))
+
+ for acc in accounts:
+ email = acc.get("email", "unknown")
+ folder = acc.get("folder", "INBOX")
+ provider = acc.get("provider", "unknown")
+ if not acc.get("password"):
+ msg = (
+ f"{provider} account {email}/{folder}: missing password/app-password - skipping"
+ )
+ logger.warning(msg)
+ diag["errors"].append(msg)
+ continue
+ diag["accounts"] += 1
+ try:
+ raw_list = cert_imap.fetch_raw_emails(
+ acc, allowlist, max_mails=per_account_max, search_days=search_days
+ )
+ except (OSError, ValueError, RuntimeError, imaplib.IMAP4.error) as e:
+ msg = f"IMAP error {email}/{folder} ({provider}): {e}"
+ logger.warning(msg)
+ diag["errors"].append(msg)
+ continue
+ diag["raw_fetched"] += len(raw_list)
+
+ for raw in raw_list:
+ try:
+ adv = parse_cert_tr_email(raw)
+ if not adv:
+ continue
+ from_hdr = adv.raw.get("from") or adv.author_login or ""
+ if allowlist and not is_cert_tr_sender(from_hdr, allowlist):
+ diag["filtered_sender"] += 1
+ continue
+ all_advs.append(adv)
+ diag["parsed"] += 1
+ except (ValueError, TypeError, AttributeError) as e:
+ logger.debug("parse_cert_tr_email failed: %s", e, exc_info=True)
+ continue
+
+ # Dedup by ghsa_id keep newest updated_at
+ seen: dict[str, NormalizedAdvisory] = {}
+ for adv in all_advs:
+ prev = seen.get(adv.ghsa_id)
+ if prev is None or adv.sort_key() > prev.sort_key():
+ seen[adv.ghsa_id] = adv
+
+ deduped = list(seen.values())
+ deduped.sort(key=lambda a: a.sort_key(), reverse=True)
+
+ logger.info(
+ "CERT-TR fetch done accounts=%d raw=%d parsed=%d deduped=%d filtered_sender=%d errors=%d",
+ diag["accounts"],
+ diag["raw_fetched"],
+ diag["parsed"],
+ len(deduped),
+ diag["filtered_sender"],
+ len(diag["errors"]),
+ )
+ return deduped, diag
diff --git a/src/advisory_rss/cli.py b/src/advisory_rss/cli.py
index 1f309cb..26d9bd8 100644
--- a/src/advisory_rss/cli.py
+++ b/src/advisory_rss/cli.py
@@ -37,13 +37,13 @@ def configure_logging(
force: bool = False,
use_stderr: bool = True,
) -> None:
- """Backwards-compat wrapper around centralized logging_config.setup_logging."""
try:
settings = get_settings()
# Prefer explicit args, else settings
lf = log_file if log_file is not None else getattr(settings, "log_file", None)
fmt = log_format if log_format != "text" else getattr(settings, "log_format", "text")
- except Exception:
+ except (OSError, ValueError, RuntimeError, AttributeError) as e:
+ logger.debug("Failed to load settings for logging config: %s", e)
lf = log_file
fmt = log_format
_setup_logging(level, log_file=lf, log_format=fmt, force=force, use_stderr=use_stderr)
@@ -60,59 +60,458 @@ def auth() -> None:
pass
+def _upsert_env_file(key: str, value: str, env_path: str = ".env") -> None:
+ import re as _re2
+
+ p = Path(env_path)
+ lines: list[str] = []
+ if p.exists():
+ try:
+ lines = p.read_text(encoding="utf-8").splitlines()
+ except OSError as e:
+ logger.debug("Failed to read %s: %s", env_path, e)
+ lines = []
+ # Find existing key
+ found = False
+ new_lines: list[str] = []
+ pattern = _re2.compile(rf"^\s*{_re2.escape(key)}\s*=", _re.IGNORECASE)
+ for line in lines:
+ if pattern.match(line):
+ if not found:
+ new_lines.append(f"{key}={value}")
+ found = True
+ # skip duplicate
+ else:
+ new_lines.append(line)
+ if not found:
+ new_lines.append(f"{key}={value}")
+ try:
+ p.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
+ logger.info("Updated %s with %s", env_path, key)
+ except OSError as e:
+ logger.warning("Failed to write %s: %s", env_path, e)
+ click.echo(f"Warning: could not write {env_path}: {e}", err=True)
+
+
+def _prompt_provider(default: str | None = None) -> str:
+ choice = click.prompt(
+ "Hangi kaynak için giriş yapmak istiyorsun?",
+ type=click.Choice(["github", "proton", "gmail"], case_sensitive=False),
+ default=default or "github",
+ show_choices=True,
+ )
+ return str(choice).lower()
+
+
+def _prompt_gmail_method(default: str | None = None) -> str:
+ choice = click.prompt(
+ "Gmail için hangi yöntem? [oauth/app-password]",
+ type=click.Choice(["oauth", "app-password", "app"], case_sensitive=False),
+ default=default or "oauth",
+ show_choices=True,
+ )
+ c = str(choice).lower()
+ return "oauth" if c == "oauth" else "app-password"
+
+
@auth.command("login")
-def auth_login() -> None:
+@click.option(
+ "--provider",
+ type=click.Choice(["github", "proton", "gmail"], case_sensitive=False),
+ default=None,
+ help="Hangi kaynak için login (github/proton/gmail). Belirtilmezse interaktif sorulur.",
+)
+@click.option(
+ "--method",
+ type=click.Choice(["oauth", "app-password"], case_sensitive=False),
+ default=None,
+ help="Gmail için yöntem (oauth veya app-password). Sadece gmail için.",
+)
+@click.option("--email", default=None, help="E-posta adresi (proton/gmail için)")
+def auth_login(
+ provider: str | None = None, method: str | None = None, email: str | None = None
+) -> None:
settings = get_settings()
_setup_logging(settings.log_level, log_file=settings.log_file, log_format=settings.log_format)
- logger.info("auth login started")
- click.echo("GitHub PAT: fine-grained 'Repository security advisories: Read' is recommended.")
- click.echo("Token must be set in .env as GITHUB_TOKEN - credentials file is no longer used.")
- existing_env = settings.token
- if existing_env:
- click.echo(f"Note: GITHUB_TOKEN already set via env ({token_preview(existing_env)}).")
- click.echo("If you want to rotate, edit .env directly: GITHUB_TOKEN=...")
+ logger.info("auth login started provider=%s method=%s email=%s", provider, method, email)
- try:
- token = getpass.getpass("Paste GitHub PAT (input hidden): ").strip()
- except (EOFError, KeyboardInterrupt):
- click.echo("\nCancelled.", err=True)
- sys.exit(1)
- if not token:
- click.echo("No token entered - abort.", err=True)
- sys.exit(1)
+ # Determine provider interactively if not provided
+ if not provider:
+ provider = _prompt_provider()
+ provider = provider.lower()
- async def _validate() -> dict:
- s = get_settings()
- cache = CacheStore(s.resolved_cache_path)
- client = GitHubClient(s, token, cache=cache)
+ if provider == "github":
+ click.echo(
+ "GitHub PAT: fine-grained 'Repository security advisories: Read' is recommended."
+ )
+ click.echo(
+ "Token must be set in .env as GITHUB_TOKEN - credentials file is no longer used."
+ )
+ existing_env = settings.token
+ if existing_env:
+ click.echo(f"Note: GITHUB_TOKEN already set via env ({token_preview(existing_env)}).")
+ click.echo("If you want to rotate, edit .env directly: GITHUB_TOKEN=...")
try:
- user = await client.get_user()
- return user
- finally:
- await client.close()
+ token = getpass.getpass("Paste GitHub PAT (input hidden): ").strip()
+ except (EOFError, KeyboardInterrupt):
+ click.echo("\nCancelled.", err=True)
+ sys.exit(1)
+ if not token:
+ click.echo("No token entered - abort.", err=True)
+ sys.exit(1)
+
+ async def _validate() -> dict:
+ s = get_settings()
+ cache = CacheStore(s.resolved_cache_path)
+ client = GitHubClient(s, token, cache=cache)
+ try:
+ user = await client.get_user()
+ return user
+ finally:
+ await client.close()
+
+ try:
+ user = asyncio.run(_validate())
+ login = user.get("login", "unknown")
+ click.echo(f"+ Validated as @{login}")
+ except AuthError as e:
+ click.echo(f"- Authentication failed: {_redact(str(e))}", err=True)
+ click.echo(
+ "Check token value, expiry, and scopes (repository_advisories:read).",
+ err=True,
+ )
+ sys.exit(1)
+ except (OSError, ValueError, RuntimeError, httpx.HTTPError) as e:
+ click.echo(f"- Validation error: {_redact(str(e))}", err=True)
+ sys.exit(1)
+
+ click.echo("")
+ click.echo("Token is VALID but NOT saved to disk by this command.")
+ if click.confirm("Token .env dosyasına yazılsın mı?", default=False):
+ _upsert_env_file("GITHUB_TOKEN", token)
+ click.echo("Wrote GITHUB_TOKEN to .env")
+ else:
+ click.echo("Set it in .env:")
+ click.echo(" echo 'GITHUB_TOKEN=YOUR_TOKEN_HERE' >> .env")
+ click.echo("Then: app sync && app serve")
+ click.echo("Never commit .env (gitignored).")
+ return
+
+ if provider == "proton":
+ # Proton Bridge: email + bridge password
+ if not email:
+ email = click.prompt(
+ "Proton e-posta adresi", type=str, default=settings.proton_bridge_email or ""
+ )
+ email = email.strip()
+ if not email or "@" not in email:
+ click.echo("Geçersiz e-posta", err=True)
+ sys.exit(1)
+ # Ask for bridge password (hidden)
+ try:
+ bridge_pw = getpass.getpass(
+ f"Proton Bridge şifresi ({email}) (Bridge > Mailbox details, NOT account password): "
+ ).strip()
+ except (EOFError, KeyboardInterrupt):
+ click.echo("\nCancelled.", err=True)
+ sys.exit(1)
+ if not bridge_pw:
+ click.echo("Şifre girilmedi - abort.", err=True)
+ sys.exit(1)
+ # Validate via IMAP
+ from advisory_rss.cert_tr.imap import connect_imap
+
+ acc = {
+ "email": email,
+ "password": bridge_pw,
+ "folder": settings.proton_imap_folder or "INBOX",
+ "host": settings.proton_bridge_host or "127.0.0.1",
+ "port": str(settings.proton_imap_port),
+ "security": settings.proton_imap_security or "STARTTLS",
+ }
+ try:
+ mail = connect_imap(acc)
+ try:
+ mail.select(acc["folder"], readonly=True)
+ # Try to list folders to verify
+ mail.list()
+ finally:
+ try:
+ mail.logout()
+ except (OSError, ValueError, RuntimeError) as e:
+ logger.debug("Proton logout failed: %s", e)
+ click.echo(
+ f"+ Proton IMAP doğrulandı: {email} -> {acc['folder']} @ {acc['host']}:{acc['port']}"
+ )
+ except (OSError, ValueError, RuntimeError) as e:
+ click.echo(f"- Proton IMAP hatası: {_redact(str(e))}", err=True)
+ click.echo(
+ "Bridge çalışıyor mu? Host/port doğru mu? Şifre Bridge şifresi mi?", err=True
+ )
+ sys.exit(1)
+
+ # Save to .env - handle multi
+ # If PROTON_BRIDGE_EMAILS already has multiple, append
+ existing_emails = settings.proton_bridge_emails
+ if existing_emails and email not in [e.strip() for e in existing_emails.split(",")]:
+ # Append to multi
+ new_emails = existing_emails.strip() + f",{email}"
+ _upsert_env_file("PROTON_BRIDGE_EMAILS", new_emails)
+ # Passwords
+ existing_pw = settings.proton_bridge_passwords or ""
+ if existing_pw.strip():
+ new_pw = existing_pw.strip() + f",{bridge_pw}"
+ else:
+ # If single was set, migrate to multi
+ single_pw = settings.proton_bridge_password or ""
+ if single_pw and single_pw.strip() != bridge_pw:
+ new_pw = f"{single_pw.strip()},{bridge_pw}"
+ else:
+ new_pw = bridge_pw
+ _upsert_env_file("PROTON_BRIDGE_PASSWORDS", new_pw)
+ click.echo(
+ f"Wrote PROTON_BRIDGE_EMAILS and PROTON_BRIDGE_PASSWORDS to .env (appended {email})"
+ )
+ else:
+ _upsert_env_file("PROTON_BRIDGE_EMAIL", email)
+ _upsert_env_file("PROTON_BRIDGE_PASSWORD", bridge_pw)
+ click.echo("Wrote PROTON_BRIDGE_EMAIL and PROTON_BRIDGE_PASSWORD to .env")
+ click.echo("ENABLE_CERT_TR=true olduğundan emin olun, sonra: app sync --source cert-tr")
+ return
+
+ if provider == "gmail":
+ # Gmail: ask oauth vs app-password
+ gmail_method = method
+ if not gmail_method:
+ gmail_method = _prompt_gmail_method()
+ gmail_method = gmail_method.lower()
+ if gmail_method not in ("oauth", "app-password", "app"):
+ gmail_method = "oauth"
+ if gmail_method == "app":
+ gmail_method = "app-password"
+
+ if not email:
+ default_email = settings.gmail_email or (
+ settings.gmail_emails.split(",")[0].strip() if settings.gmail_emails else ""
+ )
+ email = click.prompt("Gmail e-posta adresi", type=str, default=default_email)
+ email = email.strip()
+ if not email or "@" not in email:
+ click.echo("Geçersiz e-posta", err=True)
+ sys.exit(1)
+
+ if gmail_method == "app-password":
+ try:
+ app_pw = getpass.getpass(
+ f"Gmail App Password ({email}) (16 haneli, Google Hesabı > Güvenlik > 2FA > Uygulama şifreleri): "
+ ).strip()
+ except (EOFError, KeyboardInterrupt):
+ click.echo("\nCancelled.", err=True)
+ sys.exit(1)
+ if not app_pw:
+ click.echo("App Password girilmedi", err=True)
+ sys.exit(1)
+ # Validate via IMAP (Gmail SSL)
+ from advisory_rss.cert_tr.imap import connect_imap
+
+ acc = {
+ "email": email,
+ "password": app_pw,
+ "folder": settings.gmail_imap_folder or "INBOX",
+ "host": settings.gmail_imap_host or "imap.gmail.com",
+ "port": str(settings.gmail_imap_port),
+ "security": settings.gmail_imap_security or "SSL",
+ }
+ try:
+ mail = connect_imap(acc)
+ try:
+ mail.select(acc["folder"], readonly=True)
+ finally:
+ try:
+ mail.logout()
+ except (OSError, ValueError, RuntimeError) as e:
+ logger.debug("Gmail logout failed: %s", e)
+ click.echo(f"+ Gmail IMAP (App Password) doğrulandı: {email}")
+ except (OSError, ValueError, RuntimeError) as e:
+ click.echo(f"- Gmail hatası: {_redact(str(e))}", err=True)
+ click.echo(
+ "App Password doğru mu? 2FA açık mı? IMAP enabled mi? (Gmail > Ayarlar > Yönlendirme ve POP/IMAP)",
+ err=True,
+ )
+ sys.exit(1)
+
+ # Save to .env (handle multi)
+ existing_emails = settings.gmail_emails
+ if existing_emails and email not in [e.strip() for e in existing_emails.split(",")]:
+ new_emails = existing_emails.strip() + f",{email}"
+ _upsert_env_file("GMAIL_EMAILS", new_emails)
+ existing_pw = settings.gmail_app_passwords or ""
+ # App passwords may contain spaces, so we join with comma
+ if existing_pw.strip():
+ new_pw = existing_pw.strip() + f",{app_pw}"
+ else:
+ single_pw = settings.gmail_app_password or settings.gmail_password or ""
+ if single_pw and single_pw.strip() != app_pw:
+ new_pw = f"{single_pw.strip()},{app_pw}"
+ else:
+ new_pw = app_pw
+ _upsert_env_file("GMAIL_APP_PASSWORDS", new_pw)
+ click.echo(
+ f"Wrote GMAIL_EMAILS and GMAIL_APP_PASSWORDS to .env (appended {email}, spaces preserved)"
+ )
+ else:
+ _upsert_env_file("GMAIL_EMAIL", email)
+ _upsert_env_file("GMAIL_APP_PASSWORD", app_pw)
+ click.echo("Wrote GMAIL_EMAIL and GMAIL_APP_PASSWORD to .env")
+ click.echo("ENABLE_CERT_TR=true olduğundan emin olun, sonra: app sync --source cert-tr")
+ return
+
+ # OAuth path
+ # Need client_id/secret
+ client_id = settings.gmail_oauth_client_id
+ client_secret = settings.gmail_oauth_client_secret
+ if not client_id or not client_secret:
+ click.echo("Gmail OAuth için Google Cloud OAuth Client ID/Secret gerekli.")
+ click.echo(
+ "Oluştur: https://console.cloud.google.com/ -> APIs & Services -> Credentials -> Create Credentials -> OAuth Client ID (Desktop)"
+ )
+ click.echo("Redirect URI: http://localhost (InstalledAppFlow otomatik)")
+ click.echo("Scopes: https://mail.google.com/")
+ if not client_id:
+ client_id = click.prompt(
+ "OAuth Client ID", type=str, default=client_id or ""
+ ).strip()
+ if not client_secret:
+ try:
+ client_secret = getpass.getpass("OAuth Client Secret (hidden): ").strip()
+ except (EOFError, KeyboardInterrupt):
+ click.echo("\nCancelled.", err=True)
+ sys.exit(1)
+ if not client_id or not client_secret:
+ click.echo("Client ID/Secret gerekli", err=True)
+ sys.exit(1)
+ # Save to .env
+ _upsert_env_file("GMAIL_OAUTH_CLIENT_ID", client_id)
+ _upsert_env_file("GMAIL_OAUTH_CLIENT_SECRET", client_secret)
+ click.echo("Wrote GMAIL_OAUTH_CLIENT_ID/SECRET to .env")
+
+ # Run OAuth flow
+ click.echo(f"Tarayıcı açılacak, Gmail hesabı {email} ile onay verin...")
+ click.echo("Eğer tarayıcı açılmazsa, terminaldeki URL'yi manuel açın.")
+ try:
+ from advisory_rss.auth.gmail import run_gmail_oauth_flow, save_gmail_oauth_token
+ except ImportError as e:
+ click.echo(f"Gerekli paket eksik: {e}", err=True)
+ click.echo("Şunu çalıştır: pip install google-auth google-auth-oauthlib", err=True)
+ sys.exit(1)
+
+ try:
+ refresh_token, access_token, expiry = run_gmail_oauth_flow(
+ client_id, client_secret, email
+ )
+ click.echo(f"+ OAuth başarılı: refresh_token alındı ({refresh_token[:8]}...)")
+ except (OSError, ValueError, RuntimeError) as e:
+ click.echo(f"- OAuth hatası: {_redact(str(e))}", err=True)
+ sys.exit(1)
+
+ # Validate via XOAUTH2 IMAP
+ from advisory_rss.cert_tr.imap import connect_imap
+
+ acc = {
+ "email": email,
+ "password": "",
+ "folder": settings.gmail_imap_folder or "INBOX",
+ "host": settings.gmail_imap_host or "imap.gmail.com",
+ "port": str(settings.gmail_imap_port),
+ "security": settings.gmail_imap_security or "SSL",
+ "auth_method": "oauth",
+ "oauth_refresh_token": refresh_token,
+ "oauth_client_id": client_id,
+ "oauth_client_secret": client_secret,
+ }
+ try:
+ mail = connect_imap(acc)
+ try:
+ mail.select(acc["folder"], readonly=True)
+ finally:
+ try:
+ mail.logout()
+ except (OSError, ValueError, RuntimeError) as e:
+ logger.debug("Gmail OAuth logout failed: %s", e)
+ click.echo(f"+ Gmail IMAP (OAuth) doğrulandı: {email}")
+ except (OSError, ValueError, RuntimeError) as e:
+ click.echo(f"- Gmail OAuth hatası: {_redact(str(e))}", err=True)
+ sys.exit(1)
+
+ # Save refresh token
+ # Prefer file cache plus env
+ token_file = settings.gmail_oauth_token_file or "cache/gmail_oauth.json"
+ try:
+ save_gmail_oauth_token(
+ token_file, email, refresh_token, client_id, client_secret, access_token, expiry
+ )
+ click.echo(f"Wrote OAuth token to {token_file} (600 perms)")
+ except (OSError, ValueError, RuntimeError) as e:
+ logger.debug("Failed to save to file: %s", e)
+
+ # Also save to .env for convenience (single or multi)
+ existing_emails = settings.gmail_emails
+ existing_rts = settings.gmail_oauth_refresh_tokens
+ if existing_emails and email not in [e.strip() for e in existing_emails.split(",")]:
+ new_emails = existing_emails.strip() + f",{email}"
+ _upsert_env_file("GMAIL_EMAILS", new_emails)
+ if existing_rts and existing_rts.strip():
+ new_rts = existing_rts.strip() + f",{refresh_token}"
+ else:
+ single_rt = settings.gmail_oauth_refresh_token or ""
+ if single_rt and single_rt.strip() != refresh_token:
+ new_rts = f"{single_rt.strip()},{refresh_token}"
+ else:
+ new_rts = refresh_token
+ _upsert_env_file("GMAIL_OAUTH_REFRESH_TOKENS", new_rts)
+ click.echo(
+ f"Wrote GMAIL_EMAILS and GMAIL_OAUTH_REFRESH_TOKENS to .env (appended {email})"
+ )
+ elif existing_emails:
+ # Update existing entry
+ # Find index and replace
+ emails = [e.strip() for e in existing_emails.split(",")]
+ rts = [r.strip() for r in (existing_rts or "").split(",")] if existing_rts else []
+ # Pad rts
+ while len(rts) < len(emails):
+ rts.append("")
+ try:
+ idx = [e.lower() for e in emails].index(email.lower())
+ rts[idx] = refresh_token
+ _upsert_env_file("GMAIL_OAUTH_REFRESH_TOKENS", ",".join(rts))
+ click.echo("Updated GMAIL_OAUTH_REFRESH_TOKENS in .env")
+ except ValueError:
+ _upsert_env_file("GMAIL_OAUTH_REFRESH_TOKEN", refresh_token)
+ click.echo("Wrote GMAIL_OAUTH_REFRESH_TOKEN to .env")
+ else:
+ # Single
+ if settings.gmail_email and settings.gmail_email.strip().lower() != email.lower():
+ # Migrate single to multi
+ prev_email = settings.gmail_email.strip()
+ prev_rt = settings.gmail_oauth_refresh_token or ""
+ _upsert_env_file("GMAIL_EMAILS", f"{prev_email},{email}")
+ _upsert_env_file("GMAIL_OAUTH_REFRESH_TOKENS", f"{prev_rt},{refresh_token}")
+ click.echo("Migrated single Gmail to multi (GMAIL_EMAILS)")
+ else:
+ _upsert_env_file("GMAIL_EMAIL", email)
+ _upsert_env_file("GMAIL_OAUTH_REFRESH_TOKEN", refresh_token)
+ click.echo("Wrote GMAIL_EMAIL and GMAIL_OAUTH_REFRESH_TOKEN to .env")
- try:
- user = asyncio.run(_validate())
- login = user.get("login", "unknown")
- click.echo(f"+ Validated as @{login}")
- except AuthError as e:
- click.echo(f"- Authentication failed: {_redact(str(e))}", err=True)
click.echo(
- "Check token value, expiry, and scopes (repository_advisories:read).",
- err=True,
+ "OAuth tamamlandı. ENABLE_CERT_TR=true olduğundan emin olun, sonra: app sync --source cert-tr"
)
- sys.exit(1)
- except (OSError, ValueError, RuntimeError, httpx.HTTPError) as e:
- click.echo(f"- Validation error: {_redact(str(e))}", err=True)
- sys.exit(1)
+ click.echo(
+ "Not: Access token otomatik yenilenecek (refresh_token saklandı). İptal için Google Hesabı > Güvenlik > Üçüncü taraf erişimi'nden kaldırın."
+ )
+ return
- click.echo("")
- click.echo("Token is VALID but NOT saved to disk by this command.")
- click.echo("Set it in .env:")
- click.echo(" echo 'GITHUB_TOKEN=YOUR_TOKEN_HERE' >> .env")
- click.echo(" # or edit .env: GITHUB_TOKEN=github_pat_... / ghp_...")
- click.echo("Then: app sync && app serve")
- click.echo("Never commit .env (gitignored).")
+ click.echo(f"Bilinmeyen provider: {provider}", err=True)
+ sys.exit(1)
@auth.command("status")
@@ -121,90 +520,185 @@ def auth_status() -> None:
@cli.command()
-def sync() -> None:
+@click.option(
+ "--source",
+ type=click.Choice(["all", "github", "cert-tr"], case_sensitive=False),
+ default="all",
+ show_default=True,
+ help="Which source to sync (all includes GitHub + CERT-TR)",
+)
+def sync(source: str = "all") -> None:
settings = get_settings()
_setup_logging(settings.log_level, log_file=settings.log_file, log_format=settings.log_format)
logger.info(
- "sync started",
+ "sync started source=%s",
+ source,
extra={
"filter_mode": settings.filter_mode,
"cache_path": str(settings.resolved_cache_path),
+ "enable_cert_tr": settings.enable_cert_tr,
},
)
+ source = source.lower()
token = load_token(settings)
- if not token:
- click.echo("No token set. Set GITHUB_TOKEN in .env or run `app auth login`.", err=True)
- sys.exit(1)
+ # Allow cert-tr only sync without GitHub token
+ need_github = source in ("all", "github")
+ if need_github and not token:
+ # If CERT-TR is enabled and user asked all, still allow cert-tr part to run
+ if settings.enable_cert_tr and source == "all":
+ click.echo(
+ "Warning: GITHUB_TOKEN not set - GitHub sync will be skipped, CERT-TR only.",
+ err=True,
+ )
+ logger.warning("sync without GitHub token - GitHub skipped, CERT-TR only")
+ need_github = False
+ else:
+ click.echo("No token set. Set GITHUB_TOKEN in .env or run `app auth login`.", err=True)
+ sys.exit(1)
cache = CacheStore(settings.resolved_cache_path)
async def _run() -> int:
- client = GitHubClient(settings, token, cache=cache)
- try:
- advs, diag = await client.sync_all(filter_mode=settings.filter_mode)
- count = cache.upsert_advisories(advs)
- cache.mark_success(user=diag.get("authenticated_login"))
- nxt = (datetime.now(UTC) + timedelta(seconds=settings.refresh_interval)).isoformat()
- cache.set_meta("next_scheduled_sync", nxt)
- logger.info(
- "sync ok: upserted=%d filtered=%s raw=%s repos=%s orgs=%s",
- count,
- diag.get("filtered_count"),
- diag.get("raw_count"),
- diag.get("repos_scanned"),
- diag.get("orgs_scanned"),
- extra={"diag": diag},
- )
- click.echo(
- f"Sync OK - {count} advisories upserted (filtered {diag.get('filtered_count')} from {diag.get('raw_count')} raw, {diag.get('repos_scanned')} repos, {diag.get('orgs_scanned')} orgs)"
- )
- if diag.get("errors"):
- click.echo(f"Warnings ({len(diag['errors'])}):", err=True)
- for e in diag["errors"][:5]:
- click.echo(f" - {_redact(e)}", err=True)
- for e in diag["errors"]:
- logger.warning("sync warning: %s", _redact(e))
- if advs:
- click.echo(f"Latest: {advs[0].ghsa_id} - {advs[0].summary[:80]}")
- logger.debug("latest advisory: %s - %s", advs[0].ghsa_id, advs[0].summary[:80])
- else:
- click.echo(
- "No advisories matched filter - cache remains empty. Check FILTER_MODE or author login."
+ all_advisories = []
+ combined_diag: dict = {
+ "github": None,
+ "cert_tr": None,
+ "errors": [],
+ }
+ # --- GitHub ---
+ if need_github and token:
+ client = GitHubClient(settings, token, cache=cache)
+ try:
+ advs, diag = await client.sync_all(filter_mode=settings.filter_mode)
+ all_advisories.extend(advs)
+ combined_diag["github"] = diag
+ combined_diag["errors"].extend(diag.get("errors") or [])
+ logger.info(
+ "GitHub sync ok: upserted=%d filtered=%s raw=%s repos=%s orgs=%s",
+ len(advs),
+ diag.get("filtered_count"),
+ diag.get("raw_count"),
+ diag.get("repos_scanned"),
+ diag.get("orgs_scanned"),
+ extra={"diag": diag},
)
- logger.warning(
- "sync produced no advisories (filter=%s login=%s)",
- settings.filter_mode,
- diag.get("authenticated_login"),
+ click.echo(
+ f"GitHub sync OK - {len(advs)} advisories (filtered {diag.get('filtered_count')} from {diag.get('raw_count')} raw, {diag.get('repos_scanned')} repos)"
)
- return count
- except AuthError as e:
- msg = _redact(str(e))
- logger.error("sync auth error: %s", msg, exc_info=True)
- click.echo(f"Auth error (401): {msg}", err=True)
- cache.mark_error(msg)
- sys.exit(1)
- except RateLimitError as e:
- msg = _redact(str(e))
- logger.warning("sync rate limited retry_after=%s: %s", e.retry_after, msg)
+ if advs:
+ click.echo(f"Latest GitHub: {advs[0].ghsa_id} - {advs[0].summary[:80]}")
+ except AuthError as e:
+ msg = _redact(str(e))
+ logger.exception("GitHub sync auth error: %s", msg)
+ click.echo(f"GitHub Auth error (401): {msg}", err=True)
+ cache.mark_error(msg)
+ if source == "github":
+ sys.exit(1)
+ combined_diag["errors"].append(msg)
+ except RateLimitError as e:
+ msg = _redact(str(e))
+ logger.warning("GitHub rate limited retry_after=%s: %s", e.retry_after, msg)
+ click.echo(f"GitHub rate limited - retry after {e.retry_after}: {msg}", err=True)
+ if e.retry_after:
+ cache.set_meta(
+ "rate_limited_until",
+ (datetime.now(UTC) + timedelta(seconds=e.retry_after)).isoformat(),
+ )
+ cache.mark_error(msg)
+ if source == "github":
+ sys.exit(1)
+ combined_diag["errors"].append(msg)
+ except (OSError, ValueError, RuntimeError, httpx.HTTPError) as e:
+ msg = _redact(str(e))
+ logger.exception("GitHub sync failed: %s", msg)
+ click.echo(f"GitHub sync failed: {msg}", err=True)
+ cache.mark_error(msg)
+ if source == "github":
+ sys.exit(1)
+ combined_diag["errors"].append(msg)
+ finally:
+ await client.close()
+ elif source in ("all", "github"):
+ click.echo("GitHub sync skipped (no token).")
+
+ # --- CERT-TR (Proton) ---
+ if source in ("all", "cert-tr") and settings.enable_cert_tr:
+ try:
+ from advisory_rss.cert_tr.source import CertTrSource
+
+ src = CertTrSource(settings)
+ advs_ct, diag_ct = src.fetch_all()
+ combined_diag["cert_tr"] = diag_ct
+ combined_diag["errors"].extend(diag_ct.get("errors") or [])
+ if advs_ct:
+ all_advisories.extend(advs_ct)
+ click.echo(
+ f"CERT-TR sync OK - {len(advs_ct)} advisories from {diag_ct.get('accounts')} account(s) (raw {diag_ct.get('raw_fetched')})"
+ )
+ click.echo(f"Latest CERT-TR: {advs_ct[0].ghsa_id} - {advs_ct[0].summary[:80]}")
+ else:
+ click.echo(
+ f"CERT-TR sync: no advisories (raw {diag_ct.get('raw_fetched')}, filtered_sender {diag_ct.get('filtered_sender')})"
+ )
+ if diag_ct.get("errors"):
+ for e in diag_ct["errors"][:3]:
+ click.echo(f" CERT-TR warn: {_redact(e)}", err=True)
+ except (OSError, ValueError, RuntimeError) as e:
+ msg = _redact(str(e))
+ logger.exception("CERT-TR sync failed: %s", msg)
+ click.echo(f"CERT-TR sync failed: {msg}", err=True)
+ cache.mark_error(msg)
+ if source == "cert-tr":
+ sys.exit(1)
+ combined_diag["errors"].append(msg)
+ elif source in ("all", "cert-tr") and not settings.enable_cert_tr:
+ click.echo("CERT-TR disabled (ENABLE_CERT_TR=false) - skipping.", err=True)
+
+ # --- Upsert combined ---
+ if not all_advisories:
+ # Check if both sources were skipped vs empty result
click.echo(
- f"Rate limited - retry after {e.retry_after}: {msg}",
- err=True,
+ "No advisories from any source - cache unchanged. Check GITHUB_TOKEN / PROTON_* settings."
)
- if e.retry_after:
+ # Still mark success if no error (to update next sync)
+ if not combined_diag["errors"]:
+ cache.mark_success(user=cache.get_meta("authenticated_user"))
+ nxt = (datetime.now(UTC) + timedelta(seconds=settings.refresh_interval)).isoformat()
+ cache.set_meta("next_scheduled_sync", nxt)
+ else:
+ cache.mark_error("; ".join(combined_diag["errors"][:3]))
+ return 0
+
+ # Dedup across sources (GitHub GHSA-* vs CERT-TR-* no collision, but keep logic)
+ # CERT-TR already deduped by CVE; GitHub deduped too; cross-source no overlap
+ count = cache.upsert_advisories(all_advisories)
+ # keep authenticated_user from GitHub if present, else cert-tr
+ user = None
+ if combined_diag.get("github"):
+ user = combined_diag["github"].get("authenticated_login")
+ if not user:
+ user = cache.get_meta("authenticated_user")
+ cache.mark_success(user=user)
+ nxt = (datetime.now(UTC) + timedelta(seconds=settings.refresh_interval)).isoformat()
+ cache.set_meta("next_scheduled_sync", nxt)
+ # Store cert-tr diag for health
+ try:
+ import json
+
+ if combined_diag.get("cert_tr"):
cache.set_meta(
- "rate_limited_until",
- (datetime.now(UTC) + timedelta(seconds=e.retry_after)).isoformat(),
+ "cert_tr_diag", json.dumps(combined_diag["cert_tr"], ensure_ascii=False)
)
- cache.mark_error(msg)
- sys.exit(1)
- except (OSError, ValueError, RuntimeError, httpx.HTTPError) as e:
- msg = _redact(str(e))
- logger.error("sync failed: %s", msg, exc_info=True)
- click.echo(f"Sync failed: {msg}", err=True)
- cache.mark_error(msg)
- sys.exit(1)
- finally:
- await client.close()
+ except (OSError, ValueError, TypeError, RuntimeError) as e:
+ logger.debug("Failed to persist cert_tr_diag: %s", e)
+ click.echo(
+ f"Sync OK - {count} advisories upserted total ({len(all_advisories)} fetched before dedup)"
+ )
+ if combined_diag["errors"]:
+ click.echo(f"Warnings ({len(combined_diag['errors'])}):", err=True)
+ for e in combined_diag["errors"][:5]:
+ click.echo(f" - {_redact(e)}", err=True)
+ return count
asyncio.run(_run())
@@ -394,7 +888,7 @@ def serve(
except OSError as e:
logger.debug("stale pid cleanup failed: %s", _redact(str(e)))
click.echo(
- f"Daemonizing → pid: {pid_file}, log: {log_file} (terminal kapanınca da yaşar, nohup/setsid)"
+ f"Daemonizing - pid: {pid_file}, log: {log_file} (terminal kapanınca da yaşar, nohup/setsid)"
)
_daemonize(pid_file, log_file)
# Child continues; stdout/stderr now goes to log file
@@ -408,7 +902,7 @@ def serve(
use_stderr=False,
)
logger.info("daemon child logging reconfigured file=%s", log_file)
- except Exception as e:
+ except (OSError, ValueError, RuntimeError) as e:
print(f"daemon logging reconfigure failed: {e}", file=sys.stderr)
# Import uvicorn late
@@ -530,6 +1024,17 @@ def _status() -> None:
click.echo(f"Bind address: {settings.effective_bind_address()}:{settings.port}")
click.echo(f"Cache path: {settings.resolved_cache_path}")
click.echo(f"Cached advisories: {meta.advisories_count}")
+ # Per-source breakdown
+ try:
+ all_adv = cache.load_all()
+ github_c = sum(
+ 1 for a in all_adv if (getattr(a, "source", "github") or "github") == "github"
+ )
+ cert_c = sum(1 for a in all_adv if getattr(a, "source", "") == "cert-tr")
+ click.echo(f" - GitHub: {github_c}")
+ click.echo(f" - CERT-TR: {cert_c}")
+ except (OSError, ValueError, RuntimeError) as e:
+ logger.debug("Failed to load per-source breakdown: %s", e)
click.echo(f"Last sync: {meta.last_successful_sync or 'never'}")
click.echo(f"Next sync: {meta.next_scheduled_sync or 'not scheduled'}")
click.echo(f"Rate limited until: {meta.rate_limited_until or 'no'}")
@@ -539,6 +1044,23 @@ def _status() -> None:
click.echo(f"Filter mode: {settings.filter_mode}")
click.echo(f"Refresh interval: {settings.refresh_interval}s")
click.echo(f"Max items (RSS): {settings.max_items}")
+ click.echo(f"CERT-TR enabled: {settings.enable_cert_tr}")
+ if settings.enable_cert_tr:
+ accs = settings.get_cert_tr_accounts()
+ click.echo(f"CERT-TR accounts: {len(accs)}")
+ for a in accs:
+ # Never show password
+ prov = a.get("provider", "unknown")
+ click.echo(
+ f" - [{prov}] {a['email']} -> folder={a['folder']} host={a['host']}:{a['port']} sec={a['security']}"
+ )
+ click.echo(f"CERT-TR allowlist: {settings.cert_tr_sender_allowlist}")
+ click.echo(
+ f"CERT-TR max mails: {settings.cert_tr_max_mails} search_days: {settings.cert_tr_search_days or 'all'}"
+ )
+ diag_raw = cache.get_meta("cert_tr_diag")
+ if diag_raw:
+ click.echo(f"CERT-TR last diag: {diag_raw[:300]}")
# Also try to infer loopback status
from advisory_rss.server.bind import is_loopback
@@ -546,8 +1068,10 @@ def _status() -> None:
click.echo(f"Bind is loopback: {bind_ok}")
if not bind_ok:
click.echo("WARNING: bind address is NOT loopback - will refuse to start", err=True)
- if not token:
+ if not token and not settings.enable_cert_tr:
click.echo("Hint: set GITHUB_TOKEN in .env or run `app auth login`", err=True)
+ elif not token:
+ click.echo("Note: GITHUB_TOKEN not set - GitHub source disabled, CERT-TR only", err=True)
# Also support `app auth login` already; add root-level `app login` alias for convenience
diff --git a/src/advisory_rss/config/constants.py b/src/advisory_rss/config/constants.py
index da378d0..6c1704a 100644
--- a/src/advisory_rss/config/constants.py
+++ b/src/advisory_rss/config/constants.py
@@ -30,5 +30,26 @@
GITHUB_API_VERSION = "2022-11-28"
# Pinned to 2022-11-28 (stable). 2026-03-10 also valid but not required.
+# Cert-TR / Proton + Gmail
+DEFAULT_CERT_TR_ENABLED = False
+DEFAULT_PROTON_IMAP_HOST = "127.0.0.1"
+DEFAULT_PROTON_IMAP_PORT = 1143
+DEFAULT_PROTON_IMAP_SECURITY = "STARTTLS"
+DEFAULT_PROTON_FOLDER = "INBOX"
+DEFAULT_GMAIL_IMAP_HOST = "imap.gmail.com"
+DEFAULT_GMAIL_IMAP_PORT = 993
+DEFAULT_GMAIL_IMAP_SECURITY = "SSL"
+DEFAULT_GMAIL_FOLDER = "INBOX"
+DEFAULT_CERT_TR_SENDER_ALLOWLIST = "siberguvenlik.gov.tr"
+
+# Gmail OAuth
+GMAIL_OAUTH_SCOPES = ["https://mail.google.com/"]
+GMAIL_OAUTH_TOKEN_URI = "https://oauth2.googleapis.com/token"
+GMAIL_OAUTH_AUTH_URI = "https://accounts.google.com/o/oauth2/auth"
+DEFAULT_GMAIL_OAUTH_CACHE = "cache/gmail_oauth.json"
+
# Security
TOKEN_REDACT_PATTERN = r"(gh[pousr]_[A-Za-z0-9_-]+|github_pat_[A-Za-z0-9_-]+)"
+
+# Proton bridge password redaction (append to generic pattern via separate check)
+PROTON_PASSWORD_HINT = "proton"
diff --git a/src/advisory_rss/config/settings.py b/src/advisory_rss/config/settings.py
index 45d9be4..8d1097b 100644
--- a/src/advisory_rss/config/settings.py
+++ b/src/advisory_rss/config/settings.py
@@ -11,14 +11,25 @@
from advisory_rss.config.constants import (
DEFAULT_BIND_ADDRESS,
DEFAULT_CACHE_PATH,
+ DEFAULT_CERT_TR_ENABLED,
+ DEFAULT_CERT_TR_SENDER_ALLOWLIST,
DEFAULT_FILTER_MODE,
DEFAULT_GITHUB_API_BASE,
+ DEFAULT_GMAIL_FOLDER,
+ DEFAULT_GMAIL_IMAP_HOST,
+ DEFAULT_GMAIL_IMAP_PORT,
+ DEFAULT_GMAIL_IMAP_SECURITY,
+ DEFAULT_GMAIL_OAUTH_CACHE,
DEFAULT_LOG_FORMAT,
DEFAULT_LOG_LEVEL,
DEFAULT_MAX_ITEMS,
DEFAULT_MAX_PAGES_PER_REPO,
DEFAULT_MAX_REPOS,
DEFAULT_PORT,
+ DEFAULT_PROTON_FOLDER,
+ DEFAULT_PROTON_IMAP_HOST,
+ DEFAULT_PROTON_IMAP_PORT,
+ DEFAULT_PROTON_IMAP_SECURITY,
DEFAULT_REFRESH_INTERVAL,
)
from advisory_rss.server.bind import validate_bind_address
@@ -97,6 +108,73 @@ def __init__(
default=DEFAULT_MAX_PAGES_PER_REPO, validation_alias="MAX_PAGES_PER_REPO"
)
+ # Cert-TR / Proton Mail (multi-account, per-mailbox folder)
+ enable_cert_tr: bool = Field(default=DEFAULT_CERT_TR_ENABLED, validation_alias="ENABLE_CERT_TR")
+ proton_bridge_email: str | None = Field(default=None, validation_alias="PROTON_BRIDGE_EMAIL")
+ proton_bridge_password: str | None = Field(
+ default=None, validation_alias="PROTON_BRIDGE_PASSWORD"
+ )
+ proton_bridge_host: str = Field(
+ default=DEFAULT_PROTON_IMAP_HOST, validation_alias="PROTON_BRIDGE_HOST"
+ )
+ proton_imap_port: int = Field(
+ default=DEFAULT_PROTON_IMAP_PORT, validation_alias="PROTON_IMAP_PORT"
+ )
+ proton_imap_security: str = Field(
+ default=DEFAULT_PROTON_IMAP_SECURITY, validation_alias="PROTON_IMAP_SECURITY"
+ )
+ proton_imap_folder: str = Field(
+ default=DEFAULT_PROTON_FOLDER, validation_alias="PROTON_IMAP_FOLDER"
+ )
+ # Multi-account overrides (comma/space separated)
+ proton_bridge_emails: str | None = Field(default=None, validation_alias="PROTON_BRIDGE_EMAILS")
+ proton_bridge_passwords: str | None = Field(
+ default=None, validation_alias="PROTON_BRIDGE_PASSWORDS"
+ )
+ proton_imap_folders: str | None = Field(default=None, validation_alias="PROTON_IMAP_FOLDERS")
+ # Gmail (alternative IMAP for CERT-TR, e.g., imap.gmail.com)
+ gmail_email: str | None = Field(default=None, validation_alias="GMAIL_EMAIL")
+ gmail_app_password: str | None = Field(default=None, validation_alias="GMAIL_APP_PASSWORD")
+ gmail_password: str | None = Field(default=None, validation_alias="GMAIL_PASSWORD")
+ gmail_imap_host: str = Field(
+ default=DEFAULT_GMAIL_IMAP_HOST, validation_alias="GMAIL_IMAP_HOST"
+ )
+ gmail_imap_port: int = Field(
+ default=DEFAULT_GMAIL_IMAP_PORT, validation_alias="GMAIL_IMAP_PORT"
+ )
+ gmail_imap_security: str = Field(
+ default=DEFAULT_GMAIL_IMAP_SECURITY, validation_alias="GMAIL_IMAP_SECURITY"
+ )
+ gmail_imap_folder: str = Field(
+ default=DEFAULT_GMAIL_FOLDER, validation_alias="GMAIL_IMAP_FOLDER"
+ )
+ gmail_emails: str | None = Field(default=None, validation_alias="GMAIL_EMAILS")
+ gmail_app_passwords: str | None = Field(default=None, validation_alias="GMAIL_APP_PASSWORDS")
+ gmail_imap_folders: str | None = Field(default=None, validation_alias="GMAIL_IMAP_FOLDERS")
+ # Gmail OAuth (alternative to App Password, per-account refresh token)
+ gmail_oauth_client_id: str | None = Field(
+ default=None, validation_alias="GMAIL_OAUTH_CLIENT_ID"
+ )
+ gmail_oauth_client_secret: str | None = Field(
+ default=None, validation_alias="GMAIL_OAUTH_CLIENT_SECRET"
+ )
+ gmail_oauth_refresh_token: str | None = Field(
+ default=None, validation_alias="GMAIL_OAUTH_REFRESH_TOKEN"
+ )
+ gmail_oauth_refresh_tokens: str | None = Field(
+ default=None, validation_alias="GMAIL_OAUTH_REFRESH_TOKENS"
+ )
+ gmail_oauth_token_file: str = Field(
+ default=DEFAULT_GMAIL_OAUTH_CACHE, validation_alias="GMAIL_OAUTH_TOKEN_FILE"
+ )
+ # Optional: per-account JSON-like override not needed; plural fields suffice
+ cert_tr_sender_allowlist: str = Field(
+ default=DEFAULT_CERT_TR_SENDER_ALLOWLIST,
+ validation_alias="CERT_TR_SENDER_ALLOWLIST",
+ )
+ cert_tr_max_mails: int = Field(default=200, validation_alias="CERT_TR_MAX_MAILS")
+ cert_tr_search_days: int | None = Field(default=None, validation_alias="CERT_TR_SEARCH_DAYS")
+
def extra_repo_list(self) -> list[str]:
import logging
@@ -226,6 +304,105 @@ def _validate_log_format(cls, v: str) -> str:
raise ValueError(f"LOG_FORMAT must be 'text' or 'json' (got {v!r})")
return fmt
+ @field_validator("proton_imap_security")
+ @classmethod
+ def _validate_proton_security(cls, v: str) -> str:
+ lvl = (v or DEFAULT_PROTON_IMAP_SECURITY).strip().upper()
+ if lvl not in {"STARTTLS", "SSL", "NONE"}:
+ raise ValueError("PROTON_IMAP_SECURITY must be STARTTLS|SSL|NONE")
+ return lvl
+
+ @field_validator("proton_imap_folder")
+ @classmethod
+ def _validate_proton_folder(cls, v: str) -> str:
+ fv = (v or DEFAULT_PROTON_FOLDER).strip()
+ if not fv:
+ return DEFAULT_PROTON_FOLDER
+ # Allow IMAP hierarchy chars: letters, digits, /, ., -, _, space, brackets for Gmail
+ import re
+
+ if len(fv) > 100:
+ raise ValueError("PROTON_IMAP_FOLDER too long")
+ if not re.match(r"^[\w .\-/\[\]]+$", fv):
+ raise ValueError(f"PROTON_IMAP_FOLDER {fv!r} contains invalid chars")
+ return fv
+
+ @field_validator("proton_imap_folders")
+ @classmethod
+ def _validate_proton_folders(cls, v: str | None) -> str | None:
+ if v is None or not str(v).strip():
+ return v
+ import re
+
+ raw = str(v).strip()
+ if len(raw) > 500:
+ raise ValueError("PROTON_IMAP_FOLDERS too long")
+ # split only on comma/semicolon to preserve spaces/brackets inside folder names
+ parts = [p.strip() for p in re.split(r"[;,]+", raw) if p.strip()]
+ # if no comma/semicolon but spaces, fallback to comma/whitespace split for backward compat
+ if len(parts) == 1 and "," not in raw and ";" not in raw:
+ parts = [p.strip() for p in re.split(r"[,\s;]+", raw) if p.strip()]
+ for p in parts:
+ if len(p) > 100 or not re.match(r"^[\w .\-/\[\]]+$", p):
+ raise ValueError(f"PROTON_IMAP_FOLDERS entry {p!r} invalid")
+ return v
+
+ @field_validator("gmail_imap_security")
+ @classmethod
+ def _validate_gmail_security(cls, v: str) -> str:
+ lvl = (v or DEFAULT_GMAIL_IMAP_SECURITY).strip().upper()
+ if lvl not in {"STARTTLS", "SSL", "NONE"}:
+ raise ValueError("GMAIL_IMAP_SECURITY must be STARTTLS|SSL|NONE")
+ return lvl
+
+ @field_validator("gmail_imap_folder")
+ @classmethod
+ def _validate_gmail_folder(cls, v: str) -> str:
+ fv = (v or DEFAULT_GMAIL_FOLDER).strip()
+ if not fv:
+ return DEFAULT_GMAIL_FOLDER
+ import re
+
+ if len(fv) > 100:
+ raise ValueError("GMAIL_IMAP_FOLDER too long")
+ if not re.match(r"^[\w .\-/\[\]]+$", fv):
+ raise ValueError(f"GMAIL_IMAP_FOLDER {fv!r} contains invalid chars")
+ return fv
+
+ @field_validator("gmail_imap_folders")
+ @classmethod
+ def _validate_gmail_folders(cls, v: str | None) -> str | None:
+ if v is None or not str(v).strip():
+ return v
+ import re
+
+ raw = str(v).strip()
+ if len(raw) > 500:
+ raise ValueError("GMAIL_IMAP_FOLDERS too long")
+ parts = [p.strip() for p in re.split(r"[;,]+", raw) if p.strip()]
+ if len(parts) == 1 and "," not in raw and ";" not in raw:
+ parts = [p.strip() for p in re.split(r"[,\s;]+", raw) if p.strip()]
+ for p in parts:
+ if len(p) > 100 or not re.match(r"^[\w .\-/\[\]]+$", p):
+ raise ValueError(f"GMAIL_IMAP_FOLDERS entry {p!r} invalid")
+ return v
+
+ @field_validator("cert_tr_max_mails")
+ @classmethod
+ def _validate_cert_tr_max(cls, v: int) -> int:
+ if not (1 <= v <= 5000):
+ raise ValueError("CERT_TR_MAX_MAILS must be 1..5000")
+ return v
+
+ @field_validator("cert_tr_search_days")
+ @classmethod
+ def _validate_search_days(cls, v: int | None) -> int | None:
+ if v is None:
+ return None
+ if not (0 <= v <= 3650):
+ raise ValueError("CERT_TR_SEARCH_DAYS must be 0..3650 or empty (all)")
+ return v
+
@field_validator("log_file")
@classmethod
def _validate_log_file(cls, v: str | None) -> str | None:
@@ -245,6 +422,363 @@ def _validate_log_file(cls, v: str | None) -> str | None:
)
return str(v).strip()
+ def _split_list(self, raw: str | None) -> list[str]:
+ if not raw or not raw.strip():
+ return []
+ # comma, whitespace or semicolon separated, keep non-empty
+ import re
+
+ parts = re.split(r"[,\s;]+", raw.strip())
+ return [p.strip() for p in parts if p.strip()]
+
+ def _split_folders(self, raw: str | None) -> list[str]:
+ if not raw or not raw.strip():
+ return []
+ import re
+
+ # Split only on comma/semicolon to preserve spaces inside folder names (e.g., "CVE Yazışmaları")
+ parts = [p.strip() for p in re.split(r"[;,]+", raw) if p.strip()]
+ if len(parts) <= 1 and "," not in raw and ";" not in raw:
+ # Fallback for backward compat where folders were space-separated without comma
+ # Only if no comma/semicolon present, split on whitespace/comma/semicolon
+ parts = [p.strip() for p in re.split(r"[,\s;]+", raw.strip()) if p.strip()]
+ return parts
+
+ def get_proton_accounts(self) -> list[dict[str, str]]:
+ """Resolve multi-account Proton config.
+
+ Supports:
+ - Single: PROTON_BRIDGE_EMAIL + PROTON_BRIDGE_PASSWORD + PROTON_IMAP_FOLDER
+ - Multi: PROTON_BRIDGE_EMAILS= a@proton.me,b@proton.me
+ PROTON_BRIDGE_PASSWORDS= pass1,pass2
+ PROTON_IMAP_FOLDERS= INBOX,INBOX.CERT-TR
+ Indices aligned; missing values fallback to single/defaults.
+ Each account is {email, password, folder, host, port, security}.
+ Folder defaults to PROTON_IMAP_FOLDER / INBOX if not per-account.
+ """
+ emails: list[str] = []
+ passwords: list[str] = []
+ folders: list[str] = []
+
+ if self.proton_bridge_emails and self.proton_bridge_emails.strip():
+ emails = self._split_list(self.proton_bridge_emails)
+ if self.proton_bridge_passwords:
+ passwords = self._split_list(self.proton_bridge_passwords)
+ if self.proton_imap_folders:
+ folders = self._split_folders(self.proton_imap_folders)
+ elif self.proton_bridge_email and self.proton_bridge_email.strip():
+ emails = [self.proton_bridge_email.strip()]
+ if self.proton_bridge_password:
+ passwords = [self.proton_bridge_password.strip()]
+ # folder for single: use explicit plural if set else singular
+ if self.proton_imap_folders and self.proton_imap_folders.strip():
+ folders = self._split_folders(self.proton_imap_folders)
+ else:
+ folders = [self.proton_imap_folder.strip() or DEFAULT_PROTON_FOLDER]
+ else:
+ return []
+
+ # Align lengths: pad passwords/folders with defaults
+ out: list[dict[str, str]] = []
+ default_folder = (
+ self.proton_imap_folder or DEFAULT_PROTON_FOLDER
+ ).strip() or DEFAULT_PROTON_FOLDER
+ # If single password provided but multiple emails, reuse it
+ if len(passwords) == 1 and len(emails) > 1:
+ passwords = passwords * len(emails)
+ # Handle single email with multiple folders: create one account per folder
+ if len(emails) == 1 and len(folders) > 1:
+ single_email = emails[0]
+ single_pw = passwords[0] if passwords else (self.proton_bridge_password or "")
+ for folder in folders:
+ folder = folder.strip() or default_folder
+ out.append(
+ {
+ "email": single_email.strip(),
+ "password": (single_pw or "").strip(),
+ "folder": folder,
+ "host": (self.proton_bridge_host or DEFAULT_PROTON_IMAP_HOST).strip(),
+ "port": str(self.proton_imap_port),
+ "security": (self.proton_imap_security or DEFAULT_PROTON_IMAP_SECURITY)
+ .strip()
+ .upper(),
+ }
+ )
+ return out
+ for idx, email in enumerate(emails):
+ if not email or "@" not in email:
+ continue
+ pw = ""
+ if idx < len(passwords):
+ pw = passwords[idx]
+ elif passwords:
+ pw = passwords[0]
+ elif self.proton_bridge_password:
+ pw = self.proton_bridge_password
+ folder = default_folder
+ if idx < len(folders) and folders[idx]:
+ folder = folders[idx]
+ # Normalize folder: strip but keep case (IMAP case-sensitive usually)
+ folder = folder.strip() or default_folder
+ out.append(
+ {
+ "email": email.strip(),
+ "password": pw.strip() if pw else "",
+ "folder": folder,
+ "host": (self.proton_bridge_host or DEFAULT_PROTON_IMAP_HOST).strip(),
+ "port": str(self.proton_imap_port),
+ "security": (self.proton_imap_security or DEFAULT_PROTON_IMAP_SECURITY)
+ .strip()
+ .upper(),
+ }
+ )
+ return out
+
+ def _load_gmail_oauth_file_tokens(self) -> dict[str, object]:
+ """Load Gmail OAuth tokens from JSON file (cache/gmail_oauth.json).
+
+ File format: {"client_id": "...", "client_secret": "...", "accounts": {"a@gmail.com": {"refresh_token": "..."}}}
+ or flat: {"a@gmail.com": "refresh_token", ...}
+ Returns dict with keys possibly containing 'accounts' or direct email->token.
+ """
+ import json
+ import logging
+ from pathlib import Path
+
+ try:
+ path = Path(self.gmail_oauth_token_file)
+ if not path.exists():
+ return {}
+ data = json.loads(path.read_text(encoding="utf-8"))
+ return data if isinstance(data, dict) else {}
+ except (OSError, ValueError, json.JSONDecodeError) as e:
+ logging.getLogger(__name__).debug(
+ "Failed to load Gmail OAuth file %s: %s", self.gmail_oauth_token_file, e
+ )
+ return {}
+
+ def get_gmail_accounts(self) -> list[dict[str, str]]:
+ """Resolve Gmail IMAP accounts for CERT-TR.
+
+ Supports:
+ - App Password: single (GMAIL_EMAIL + GMAIL_APP_PASSWORD) and
+ multi (GMAIL_EMAILS, GMAIL_APP_PASSWORDS comma-separated,
+ GMAIL_IMAP_FOLDERS). Host/port/security are shared per Gmail
+ config (defaults imap.gmail.com:993 SSL).
+ - OAuth: GMAIL_OAUTH_CLIENT_ID/SECRET + GMAIL_OAUTH_REFRESH_TOKEN(S)
+ or cache/gmail_oauth.json. For each email, if a refresh token
+ exists (env or file), the account will use XOAUTH2 (auth_method=oath).
+ Otherwise App Password is used.
+ Each account is {email,password,folder,host,port,security,provider,auth_method,
+ oauth_client_id,oauth_client_secret,oauth_refresh_token}.
+ """
+ emails: list[str] = []
+ passwords: list[str] = []
+ folders: list[str] = []
+ oauth_refresh_tokens: list[str] = []
+
+ def _split_passwords(raw: str | None) -> list[str]:
+ if not raw or not raw.strip():
+ return []
+ parts = [p.strip() for p in raw.split(",")]
+ return [p for p in parts if p]
+
+ # Determine emails source
+ if self.gmail_emails and self.gmail_emails.strip():
+ emails = self._split_list(self.gmail_emails)
+ if self.gmail_app_passwords:
+ passwords = _split_passwords(self.gmail_app_passwords)
+ if self.gmail_oauth_refresh_tokens:
+ oauth_refresh_tokens = _split_passwords(self.gmail_oauth_refresh_tokens)
+ elif self.gmail_oauth_refresh_token:
+ oauth_refresh_tokens = [self.gmail_oauth_refresh_token.strip()]
+ if self.gmail_imap_folders:
+ folders = self._split_folders(self.gmail_imap_folders)
+ elif self.gmail_email and self.gmail_email.strip():
+ emails = [self.gmail_email.strip()]
+ pw = self.gmail_app_password or self.gmail_password
+ if pw:
+ passwords = [pw.strip()]
+ # OAuth single
+ if self.gmail_oauth_refresh_tokens and self.gmail_oauth_refresh_tokens.strip():
+ oauth_refresh_tokens = _split_passwords(self.gmail_oauth_refresh_tokens)
+ elif self.gmail_oauth_refresh_token:
+ oauth_refresh_tokens = [self.gmail_oauth_refresh_token.strip()]
+ if self.gmail_imap_folders and self.gmail_imap_folders.strip():
+ folders = self._split_folders(self.gmail_imap_folders)
+ else:
+ folders = [self.gmail_imap_folder.strip() or DEFAULT_GMAIL_FOLDER]
+ else:
+ # Check OAuth file for accounts even if no env email
+ file_tokens = self._load_gmail_oauth_file_tokens()
+ # file may contain {"accounts": {"a@gmail.com": {"refresh_token": "..."}}}
+ accounts_map: dict[str, str] = {}
+ if isinstance(file_tokens, dict):
+ if "accounts" in file_tokens and isinstance(file_tokens["accounts"], dict):
+ for em, info in file_tokens["accounts"].items():
+ if isinstance(info, dict) and info.get("refresh_token"):
+ accounts_map[em] = str(info["refresh_token"])
+ elif isinstance(info, str):
+ accounts_map[em] = info
+ else:
+ # flat email->token
+ for k, v in file_tokens.items():
+ if "@" in k and isinstance(v, str):
+ accounts_map[k] = v
+ elif "@" in k and isinstance(v, dict) and v.get("refresh_token"):
+ accounts_map[k] = str(v["refresh_token"])
+ if accounts_map:
+ emails = list(accounts_map.keys())
+ oauth_refresh_tokens = [accounts_map[e] for e in emails]
+ # folders will be default
+ else:
+ return []
+
+ # Also check file for additional OAuth tokens that may supplement env
+ file_tokens = self._load_gmail_oauth_file_tokens()
+ file_map: dict[str, str] = {}
+ if isinstance(file_tokens, dict):
+ if "accounts" in file_tokens and isinstance(file_tokens["accounts"], dict):
+ for em, info in file_tokens["accounts"].items():
+ if isinstance(info, dict) and info.get("refresh_token"):
+ file_map[em.lower()] = str(info["refresh_token"])
+ elif isinstance(info, str):
+ file_map[em.lower()] = info
+ else:
+ for k, v in file_tokens.items():
+ if "@" in k:
+ if isinstance(v, str):
+ file_map[k.lower()] = v
+ elif isinstance(v, dict) and v.get("refresh_token"):
+ file_map[k.lower()] = str(v["refresh_token"])
+ # Also consider env file client_id/secret for file-based accounts
+ out: list[dict[str, str]] = []
+ default_folder = (
+ self.gmail_imap_folder or DEFAULT_GMAIL_FOLDER
+ ).strip() or DEFAULT_GMAIL_FOLDER
+ if len(passwords) == 1 and len(emails) > 1:
+ passwords = passwords * len(emails)
+ if len(oauth_refresh_tokens) == 1 and len(emails) > 1:
+ oauth_refresh_tokens = oauth_refresh_tokens * len(emails)
+ # Handle single email with multiple folders: expand to multiple accounts
+ if len(emails) == 1 and len(folders) > 1:
+ single_email = emails[0]
+ single_pw = (
+ passwords[0]
+ if passwords
+ else (self.gmail_app_password or self.gmail_password or "")
+ )
+ single_oauth = oauth_refresh_tokens[0] if oauth_refresh_tokens else ""
+ for folder in folders:
+ folder = folder.strip() or default_folder
+ # Determine OAuth token for this folder (same for all folders of same email)
+ oauth_rt = single_oauth
+ # Also check file_map for this email
+ # file_map will be computed later, but for now handle simple
+ auth_method = "oauth" if oauth_rt else "password"
+ acct: dict[str, str] = {
+ "email": single_email.strip(),
+ "password": (single_pw or "").strip(),
+ "folder": folder,
+ "host": (self.gmail_imap_host or DEFAULT_GMAIL_IMAP_HOST).strip(),
+ "port": str(self.gmail_imap_port),
+ "security": (self.gmail_imap_security or DEFAULT_GMAIL_IMAP_SECURITY)
+ .strip()
+ .upper(),
+ "provider": "gmail",
+ "auth_method": auth_method,
+ }
+ if oauth_rt:
+ acct["oauth_refresh_token"] = oauth_rt.strip()
+ cid = self.gmail_oauth_client_id or ""
+ csec = self.gmail_oauth_client_secret or ""
+ if not cid and isinstance(file_tokens, dict) and file_tokens.get("client_id"):
+ cid = str(file_tokens["client_id"])
+ if (
+ not csec
+ and isinstance(file_tokens, dict)
+ and file_tokens.get("client_secret")
+ ):
+ csec = str(file_tokens["client_secret"])
+ acct["oauth_client_id"] = (cid or "").strip()
+ acct["oauth_client_secret"] = (csec or "").strip()
+ acct["oauth_token_file"] = (self.gmail_oauth_token_file or "").strip()
+ out.append(acct)
+ return out
+ for idx, email in enumerate(emails):
+ if not email or "@" not in email:
+ continue
+ pw = ""
+ if idx < len(passwords):
+ pw = passwords[idx]
+ elif passwords:
+ pw = passwords[0]
+ else:
+ single_pw = self.gmail_app_password or self.gmail_password
+ pw = single_pw or ""
+ # Determine OAuth token for this email (env takes precedence, then file)
+ oauth_rt = ""
+ if idx < len(oauth_refresh_tokens) and oauth_refresh_tokens[idx]:
+ oauth_rt = oauth_refresh_tokens[idx]
+ elif file_map.get(email.lower()):
+ oauth_rt = file_map[email.lower()]
+ elif self.gmail_oauth_refresh_token and len(emails) == 1:
+ oauth_rt = self.gmail_oauth_refresh_token
+ folder = default_folder
+ if idx < len(folders) and folders[idx]:
+ folder = folders[idx]
+ folder = folder.strip() or default_folder
+ # Decide auth method: prefer OAuth if refresh token present
+ auth_method = "oauth" if oauth_rt else "password"
+ # For OAuth, password is not used, but keep for fallback
+ acct: dict[str, str] = {
+ "email": email.strip(),
+ "password": pw.strip() if pw else "",
+ "folder": folder,
+ "host": (self.gmail_imap_host or DEFAULT_GMAIL_IMAP_HOST).strip(),
+ "port": str(self.gmail_imap_port),
+ "security": (self.gmail_imap_security or DEFAULT_GMAIL_IMAP_SECURITY)
+ .strip()
+ .upper(),
+ "provider": "gmail",
+ "auth_method": auth_method,
+ }
+ if oauth_rt:
+ acct["oauth_refresh_token"] = oauth_rt.strip()
+ # client_id/secret may be in env or file
+ cid = self.gmail_oauth_client_id or ""
+ csec = self.gmail_oauth_client_secret or ""
+ # also check file for client_id/secret
+ if not cid and isinstance(file_tokens, dict) and file_tokens.get("client_id"):
+ cid = str(file_tokens["client_id"])
+ if not csec and isinstance(file_tokens, dict) and file_tokens.get("client_secret"):
+ csec = str(file_tokens["client_secret"])
+ acct["oauth_client_id"] = (cid or "").strip()
+ acct["oauth_client_secret"] = (csec or "").strip()
+ acct["oauth_token_file"] = (self.gmail_oauth_token_file or "").strip()
+ out.append(acct)
+ return out
+
+ def get_cert_tr_accounts(self) -> list[dict[str, str]]:
+ """Unified CERT-TR IMAP accounts: Proton + Gmail.
+
+ Each entry is {email,password,folder,host,port,security,provider}.
+ Proton entries have provider 'proton', Gmail 'gmail'.
+ """
+ proton = self.get_proton_accounts()
+ for p in proton:
+ p.setdefault("provider", "proton")
+ gmail = self.get_gmail_accounts()
+ # gmail already has provider
+ return proton + gmail
+
+ def cert_tr_sender_list(self) -> list[str]:
+ if not self.cert_tr_sender_allowlist or not self.cert_tr_sender_allowlist.strip():
+ return [DEFAULT_CERT_TR_SENDER_ALLOWLIST]
+ return [
+ s.strip().lower() for s in self._split_list(self.cert_tr_sender_allowlist) if s.strip()
+ ]
+
def effective_bind_address(self) -> str:
if self.host and self.bind_address == DEFAULT_BIND_ADDRESS:
validate_bind_address(self.host)
diff --git a/src/advisory_rss/logging_config.py b/src/advisory_rss/logging_config.py
index bcacf40..597cf32 100644
--- a/src/advisory_rss/logging_config.py
+++ b/src/advisory_rss/logging_config.py
@@ -18,6 +18,8 @@
_TOKEN_RE = re.compile(TOKEN_REDACT_PATTERN)
+logger = logging.getLogger(__name__)
+
def _redact(s: str) -> str:
return _TOKEN_RE.sub("***", s)
@@ -142,7 +144,8 @@ def format(self, record: logging.LogRecord) -> str:
extras[k] = _redact(v)
else:
extras[k] = v
- except Exception:
+ except (AttributeError, TypeError, ValueError, RuntimeError) as e:
+ logger.debug("Redact extra %s failed: %s", k, e)
extras[k] = str(v)
if extras:
payload["extra"] = extras
@@ -226,8 +229,8 @@ def setup_logging(
root.removeHandler(h)
try:
h.close()
- except Exception:
- pass
+ except (OSError, ValueError, RuntimeError) as e:
+ logger.debug("Handler close failed: %s", e)
# Choose formatter
if fmt_str == "json":
@@ -286,6 +289,7 @@ def setup_logging(
_configured_format = fmt_str
_configured_log_file = normalized_log_file or ""
+
def get_uvicorn_log_config(level: str = "INFO") -> dict[str, Any]:
lvl = (level or "INFO").upper()
if lvl not in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
@@ -341,5 +345,5 @@ def reset_logging_for_tests() -> None:
root.removeHandler(h)
try:
h.close()
- except Exception:
- pass
+ except (OSError, ValueError, RuntimeError) as e:
+ logger.debug("Handler close failed in reset: %s", e)
diff --git a/src/advisory_rss/rss/builder.py b/src/advisory_rss/rss/builder.py
index 317e9be..c114f92 100644
--- a/src/advisory_rss/rss/builder.py
+++ b/src/advisory_rss/rss/builder.py
@@ -63,9 +63,12 @@ def _build_item_xml(adv: NormalizedAdvisory) -> str:
# Title logic
severity = _severity_label(adv.severity)
is_withdrawn = bool(adv.withdrawn_at) or adv.state.lower() == "withdrawn"
+ src = getattr(adv, "source", "github") or "github"
prefix = ""
if is_withdrawn:
prefix = "[WITHDRAWN] "
+ elif src == "cert-tr":
+ prefix = "[CERT-TR] "
elif severity not in ("unknown", "n/a"):
prefix = f"[{severity.upper()}] "
base_title = adv.summary.strip() or adv.ghsa_id
@@ -73,7 +76,18 @@ def _build_item_xml(adv: NormalizedAdvisory) -> str:
title_esc = escape_text(title)
raw_link = adv.html_url.strip() if adv.html_url else ""
- link = raw_link if _is_safe_url(raw_link) else f"https://github.com/advisories/{adv.ghsa_id}"
+ if src == "cert-tr":
+ # fallback to NVD or siberguvenlik portal if not safe
+ fallback = (
+ f"https://nvd.nist.gov/vuln/detail/{adv.cve_id}"
+ if adv.cve_id
+ else "https://siberguvenlik.gov.tr"
+ )
+ link = raw_link if _is_safe_url(raw_link) else fallback
+ else:
+ link = (
+ raw_link if _is_safe_url(raw_link) else f"https://github.com/advisories/{adv.ghsa_id}"
+ )
link_esc = escape_text(link)
guid = escape_text(adv.ghsa_id)
@@ -85,6 +99,7 @@ def _build_item_xml(adv: NormalizedAdvisory) -> str:
author = escape_text(adv.author_login or "unknown")
# categories
cats: list[str] = []
+ cats.append(f"source:{src}")
cats.append(f"severity:{severity}")
if adv.package_ecosystem:
cats.append(f"ecosystem:{adv.package_ecosystem}")
@@ -97,6 +112,15 @@ def _build_item_xml(adv: NormalizedAdvisory) -> str:
# also cve if present
if adv.cve_id:
cats.append(f"cve:{adv.cve_id}")
+ # extra cves for cert-tr
+ extra = getattr(adv, "extra_cves", []) or []
+ for ec in extra[:5]:
+ cats.append(f"cve:{ec}")
+ # CWE identifier if present
+ for ident in getattr(adv, "identifiers", []) or []:
+ if isinstance(ident, dict) and ident.get("type") == "CWE":
+ cats.append(f"cwe:{ident.get('value')}")
+ break
cat_xml = "\n".join(f"
{escape_text(c)}" for c in cats)
@@ -239,7 +263,14 @@ def _build_content_html(adv: NormalizedAdvisory) -> str:
parts.append("")
view_link = adv.html_url if _is_safe_url(adv.html_url) else "#"
- parts.append(f'
View on GitHub
')
+ label = "View NVD entry" if getattr(adv, "source", "github") == "cert-tr" else "View on GitHub"
+ parts.append(f'
{label}
')
+ # source footer
+ src = getattr(adv, "source", "github") or "github"
+ if src == "cert-tr":
+ parts.append(
+ "
Source: CERT-TR / Siber Güvenlik Başkanlığı (via Proton Mail)
"
+ )
html_str = "\n".join(parts)
# Bound content size per item
diff --git a/src/advisory_rss/server/app.py b/src/advisory_rss/server/app.py
index bb6564a..6551593 100644
--- a/src/advisory_rss/server/app.py
+++ b/src/advisory_rss/server/app.py
@@ -112,10 +112,38 @@ def create_app(
@app.get("/health")
async def health() -> JSONResponse:
meta = cache.get_cache_meta()
+ # per-source counts
+ try:
+ all_advs = cache.load_all()
+ github_count = sum(
+ 1 for a in all_advs if (getattr(a, "source", "github") or "github") == "github"
+ )
+ cert_tr_count = sum(1 for a in all_advs if getattr(a, "source", "") == "cert-tr")
+ except (OSError, ValueError, RuntimeError, AttributeError) as e:
+ logger.debug("Failed to load per-source counts: %s", e)
+ github_count = None
+ cert_tr_count = None
+ # cert_tr diag
+ cert_tr_diag = None
+ try:
+ import json
+
+ raw = cache.get_meta("cert_tr_diag")
+ if raw:
+ cert_tr_diag = json.loads(raw)
+ except (OSError, ValueError, TypeError, RuntimeError) as e:
+ logger.debug("Failed to load cert_tr_diag: %s", e)
return JSONResponse(
content={
"status": "ok" if not meta.rate_limited_until else "rate_limited",
"advisories_count": meta.advisories_count,
+ "github_count": github_count,
+ "cert_tr_count": cert_tr_count,
+ "cert_tr_enabled": settings.enable_cert_tr,
+ "cert_tr_accounts": len(settings.get_cert_tr_accounts())
+ if settings.enable_cert_tr
+ else 0,
+ "cert_tr_diag": cert_tr_diag,
"last_successful_sync": meta.last_successful_sync,
"next_scheduled_sync": meta.next_scheduled_sync,
"rate_limited_until": meta.rate_limited_until,
@@ -134,26 +162,57 @@ async def health() -> JSONResponse:
@app.get("/rss.xml")
@app.get("/rss")
async def rss(request: Request) -> Response:
- if request.query_params:
+ # Optional source filter: ?source=all|github|cert-tr (default all = mixed)
+ src_filter = (request.query_params.get("source") or "all").lower().strip()
+ if src_filter not in ("all", "github", "cert-tr", "cert_tr"):
+ src_filter = "all"
+ if src_filter == "cert_tr":
+ src_filter = "cert-tr"
+ if request.query_params and "source" not in request.query_params:
logger.debug(
- "RSS query params ignored: %s path=%s",
+ "RSS query params ignored (no source): %s path=%s",
_redact(str(request.query_params)),
request.url.path,
)
logger.debug(
- "RSS request path=%s user_agent=%s",
+ "RSS request path=%s source=%s user_agent=%s",
request.url.path,
+ src_filter,
_redact(request.headers.get("user-agent", "")[:200]),
)
- # Serve from cache only - never call GitHub here
+ # Serve from cache only - never call GitHub/IMAP here
advisories = cache.load_sorted(limit=None) # load all sorted, builder caps by max_items
+ if src_filter != "all":
+ advisories = [
+ a for a in advisories if (getattr(a, "source", "github") or "github") == src_filter
+ ]
feed_user = cache.get_meta("authenticated_user") or "user"
ttl_minutes = max(1, settings.refresh_interval // 60)
+ # Mixed feed title when both sources present
+ has_github = any(
+ (getattr(a, "source", "github") or "github") == "github" for a in advisories
+ )
+ has_cert = any(getattr(a, "source", "") == "cert-tr" for a in advisories)
+ if has_github and has_cert and src_filter == "all":
+ feed_title = f"Security Advisories (GitHub + CERT-TR) - @{feed_user}"
+ feed_desc = f"Mixed feed: GitHub advisories by @{feed_user} + CERT-TR via Proton Mail"
+ elif src_filter == "cert-tr":
+ feed_title = "CERT-TR / Siber Güvenlik Başkanlığı Advisories"
+ feed_desc = "CERT-TR advisories via Proton Mail (siberguvenlik.gov.tr)"
+ else:
+ feed_title = f"GitHub Security Advisories - @{feed_user}"
+ feed_desc = f"Security advisories created by {feed_user} (author={feed_user}) via GitHub API - localhost-only feed"
+ # Keep feed_link sensible
+ feed_link = (
+ f"https://github.com/{feed_user}"
+ if src_filter != "cert-tr"
+ else "https://siberguvenlik.gov.tr"
+ )
xml = build_rss(
advisories,
- feed_title=f"GitHub Security Advisories - @{feed_user}",
- feed_link=f"https://github.com/{feed_user}",
- feed_description=f"Security advisories created by {feed_user} (author={feed_user}) via GitHub API - localhost-only feed",
+ feed_title=feed_title,
+ feed_link=feed_link,
+ feed_description=feed_desc,
max_items=settings.max_items,
ttl_minutes=ttl_minutes,
authenticated_user=feed_user,
@@ -211,35 +270,73 @@ async def refresh(request: Request) -> JSONResponse:
raise HTTPException(status_code=403, detail="Invalid refresh token")
# Enforce streaming body size even for chunked (defense in depth)
await _read_limited_body(request, MAX_POST_BYTES)
+ # Optional ?source= filter
+ q_source = (request.query_params.get("source") or "all").lower().strip()
+ if q_source not in ("all", "github", "cert-tr", "cert_tr"):
+ q_source = "all"
+ if q_source == "cert_tr":
+ q_source = "cert-tr"
# Trigger sync background
token = load_token(settings)
- if not token:
+ # Refresh allows mixed: if no GitHub token but cert-tr enabled, still refresh cert-tr
+ need_github = q_source in ("all", "github")
+ if need_github and not token and settings.enable_cert_tr and q_source == "all":
+ need_github = False
+ elif need_github and not token:
raise HTTPException(status_code=401, detail="GitHub token not configured")
# Run sync inline but bounded
try:
- client = GitHubClient(settings, token, cache=cache)
- try:
- advs, diag = await client.sync_all(filter_mode=settings.filter_mode)
- count = cache.upsert_advisories(advs)
- cache.mark_success(user=diag.get("authenticated_login"))
- # next sync
- nxt = (datetime.now(UTC) + timedelta(seconds=settings.refresh_interval)).isoformat()
- cache.set_meta("next_scheduled_sync", nxt)
- logger.info(
- "POST /refresh ok: count=%d login=%s",
- count,
- diag.get("authenticated_login"),
- extra={"diag": diag},
- )
- return JSONResponse(
- content={
- "status": "refreshed",
- "count": count,
- "diag": {k: v for k, v in diag.items() if k != "errors" or v},
- }
+ all_advs = []
+ diag_comb: dict[str, Any] = {"errors": []}
+ if need_github and token:
+ client = GitHubClient(settings, token, cache=cache)
+ try:
+ advs, diag = await client.sync_all(filter_mode=settings.filter_mode)
+ all_advs.extend(advs)
+ diag_comb["github"] = diag
+ diag_comb["errors"].extend(diag.get("errors") or [])
+ finally:
+ await client.close()
+ if q_source in ("all", "cert-tr") and settings.enable_cert_tr:
+ from advisory_rss.cert_tr.source import CertTrSource
+
+ src = CertTrSource(settings)
+ advs_ct, diag_ct = src.fetch_all()
+ all_advs.extend(advs_ct)
+ diag_comb["cert_tr"] = diag_ct
+ diag_comb["errors"].extend(diag_ct.get("errors") or [])
+ if not all_advs:
+ raise HTTPException(
+ status_code=502,
+ detail=f"Refresh: no advisories (errors={diag_comb['errors'][:2]})",
)
- finally:
- await client.close()
+ count = cache.upsert_advisories(all_advs)
+ # keep user from GitHub if present
+ user = None
+ if diag_comb.get("github"):
+ user = diag_comb["github"].get("authenticated_login")
+ cache.mark_success(user=user)
+ nxt = (datetime.now(UTC) + timedelta(seconds=settings.refresh_interval)).isoformat()
+ cache.set_meta("next_scheduled_sync", nxt)
+ try:
+ import json
+
+ if diag_comb.get("cert_tr"):
+ cache.set_meta(
+ "cert_tr_diag", json.dumps(diag_comb["cert_tr"], ensure_ascii=False)
+ )
+ except (OSError, ValueError, TypeError, RuntimeError) as e:
+ logger.debug("Failed to persist cert_tr_diag: %s", e)
+ logger.info("POST /refresh ok: count=%d", count, extra={"diag": diag_comb})
+ return JSONResponse(
+ content={
+ "status": "refreshed",
+ "count": count,
+ "diag": {k: v for k, v in diag_comb.items() if k != "errors" or v},
+ }
+ )
+ except HTTPException:
+ raise
except (OSError, ValueError, RuntimeError) as e:
msg = _redact(str(e))
logger.warning("Refresh failed: %s", msg, exc_info=True)
@@ -253,7 +350,6 @@ async def unhandled(request: Request, exc: Exception):
"Unhandled error on %s: %s",
request.url.path,
_redact(str(exc)),
- exc_info=True,
)
return JSONResponse(
status_code=500,
@@ -266,10 +362,12 @@ async def unhandled(request: Request, exc: Exception):
async def background_refresh_loop(settings: Settings, cache: CacheStore) -> None:
# Run an immediate refresh on startup (after short grace) then periodic
+ # Mixed: GitHub + CERT-TR (IMAP is sync, wrap in thread)
logger.info(
- "background_refresh_loop starting interval=%ds filter=%s",
+ "background_refresh_loop starting interval=%ds filter=%s cert_tr=%s",
settings.refresh_interval,
settings.filter_mode,
+ settings.enable_cert_tr,
)
first = True
while True:
@@ -280,25 +378,90 @@ async def background_refresh_loop(settings: Settings, cache: CacheStore) -> None
await asyncio.sleep(2)
first = False
token = load_token(settings)
- if not token:
- logger.info("Background refresh skipped - no token")
+ # If neither source is usable, skip
+ if not token and not settings.enable_cert_tr:
+ logger.info("Background refresh skipped - no token and CERT-TR disabled")
+ continue
+ # Also skip if CERT-TR enabled but no IMAP accounts and no token => nothing to do
+ if not token and settings.enable_cert_tr and not settings.get_cert_tr_accounts():
+ logger.info(
+ "Background refresh skipped - CERT-TR enabled but no IMAP accounts and no GitHub token"
+ )
continue
try:
- client = GitHubClient(settings, token, cache=cache)
- try:
- advs, diag = await client.sync_all(filter_mode=settings.filter_mode)
- cache.upsert_advisories(advs)
- cache.mark_success(user=diag.get("authenticated_login"))
- nxt = (datetime.now(UTC) + timedelta(seconds=settings.refresh_interval)).isoformat()
- cache.set_meta("next_scheduled_sync", nxt)
+ all_advs = []
+ github_diag = None
+ cert_diag = None
+ errors: list[str] = []
+ # GitHub (async)
+ if token:
+ client = GitHubClient(settings, token, cache=cache)
+ try:
+ advs, diag = await client.sync_all(filter_mode=settings.filter_mode)
+ all_advs.extend(advs)
+ github_diag = diag
+ errors.extend(diag.get("errors") or [])
+ except (OSError, ValueError, RuntimeError) as e:
+ msg = _redact(str(e))
+ logger.warning("Background GitHub refresh failed: %s", msg, exc_info=True)
+ errors.append(msg)
+ # keep stale, don't abort cert-tr
+ finally:
+ await client.close()
+ # CERT-TR (sync IMAP -> run in thread)
+ if settings.enable_cert_tr:
+ try:
+ from advisory_rss.cert_tr.source import CertTrSource
+
+ def _fetch_ct():
+ src = CertTrSource(settings)
+ return src.fetch_all()
+
+ advs_ct, diag_ct = await asyncio.to_thread(_fetch_ct)
+ cert_diag = diag_ct
+ errors.extend(diag_ct.get("errors") or [])
+ if advs_ct:
+ all_advs.extend(advs_ct)
+ except (OSError, ValueError, RuntimeError) as e:
+ msg = _redact(str(e))
+ logger.warning("Background CERT-TR refresh failed: %s", msg, exc_info=True)
+ errors.append(msg)
+ if not all_advs and errors:
+ # Report but keep stale
+ msg = "; ".join(errors[:3])
+ cache.mark_error(msg)
+ logger.warning("Background refresh: no advisories, errors=%s", msg)
+ elif all_advs:
+ # Dedup already done per source; final merge dedup by ghsa_id
+ cache.upsert_advisories(all_advs)
+ user = (github_diag or {}).get("authenticated_login") if github_diag else None
+ cache.mark_success(user=user or cache.get_meta("authenticated_user"))
+ try:
+ import json
+
+ if cert_diag is not None:
+ cache.set_meta("cert_tr_diag", json.dumps(cert_diag, ensure_ascii=False))
+ except (OSError, ValueError, TypeError, RuntimeError) as e:
+ logger.debug("Failed to persist cert_diag in background: %s", e)
+ gh_count = 0
+ if isinstance(github_diag, dict):
+ try:
+ gh_count = int(github_diag.get("filtered_count") or 0)
+ except (ValueError, TypeError) as e:
+ logger.debug("Failed to parse gh_count: %s", e)
+ gh_count = 0
+ ct_count = int(cert_diag.get("parsed") or 0) if isinstance(cert_diag, dict) else 0
logger.info(
- "Background refresh OK: %d advisories login=%s",
- len(advs),
- diag.get("authenticated_login"),
- extra={"diag": diag},
+ "Background refresh OK: total=%d github=%d cert_tr=%d errors=%d",
+ len(all_advs),
+ gh_count,
+ ct_count,
+ len(errors),
)
- finally:
- await client.close()
+ else:
+ logger.info("Background refresh: no new advisories")
+ nxt = (datetime.now(UTC) + timedelta(seconds=settings.refresh_interval)).isoformat()
+ cache.set_meta("next_scheduled_sync", nxt)
except (OSError, ValueError, RuntimeError) as e:
msg = _redact(str(e))
logger.warning("Background refresh failed, keeping stale cache: %s", msg, exc_info=True)
diff --git a/tests/unit/test_cert_tr_imap.py b/tests/unit/test_cert_tr_imap.py
new file mode 100644
index 0000000..310fba8
--- /dev/null
+++ b/tests/unit/test_cert_tr_imap.py
@@ -0,0 +1,125 @@
+import imaplib
+from unittest.mock import MagicMock, patch
+
+from advisory_rss.cert_tr.imap import fetch_raw_emails
+
+SAMPLE_RAW = (
+ b"From: cve@siberguvenlik.gov.tr\r\nSubject: Test CVE-2026-0001\r\n\r\nBody CVE-2026-0001"
+)
+
+
+def _mock_imap(data_bytes_list):
+ """Create mock IMAP object that returns uids and fetches."""
+ mock = MagicMock()
+ # select
+ mock.select.return_value = ("OK", [b"1"])
+ # search returns uids
+ mock.search.return_value = ("OK", [b"1 2"])
+
+ # fetch returns tuple per uid
+ def fetch_side(uid, _what):
+ # uid can be bytes like b"2" or b"1"
+ if isinstance(uid, bytes):
+ uid_s = uid.decode()
+ else:
+ uid_s = str(uid)
+ # map uid to sample
+ idx = 0 if uid_s == "2" else 1
+ raw = data_bytes_list[idx] if idx < len(data_bytes_list) else data_bytes_list[0]
+ return ("OK", [(b"1 (BODY[] {10}", raw)])
+
+ mock.fetch.side_effect = fetch_side
+ return mock
+
+
+def test_fetch_never_marks_read():
+ acc = {
+ "email": "a@proton.me",
+ "password": "p",
+ "folder": "INBOX",
+ "host": "127.0.0.1",
+ "port": "1143",
+ "security": "STARTTLS",
+ }
+ with patch("advisory_rss.cert_tr.imap.connect_imap") as mock_connect:
+ mock = _mock_imap([SAMPLE_RAW, SAMPLE_RAW])
+ mock_connect.return_value = mock
+ raws = fetch_raw_emails(acc, ["siberguvenlik.gov.tr"], max_mails=2)
+ # Should have used BODY.PEEK[]
+ assert mock.fetch.called
+ for call_args in mock.fetch.call_args_list:
+ args, _kwargs = call_args
+ assert "BODY.PEEK" in args[1]
+ # Should have selected readonly=True
+ assert mock.select.called
+ for c in mock.select.call_args_list:
+ args, kwargs = c
+ assert kwargs.get("readonly") is True
+ assert len(raws) == 2
+
+
+def test_fetch_multi_account_distinct_folders():
+ from advisory_rss.config.settings import Settings
+
+ s = Settings(
+ _env_file="/dev/null",
+ ENABLE_CERT_TR="true",
+ PROTON_BRIDGE_EMAILS="a@proton.me,b@proton.me",
+ PROTON_BRIDGE_PASSWORDS="p1,p2",
+ PROTON_IMAP_FOLDERS="INBOX,INBOX.CERT-TR",
+ _env_file_encoding="utf-8",
+ )
+ accs = s.get_proton_accounts()
+ assert accs[0]["folder"] == "INBOX"
+ assert accs[1]["folder"] == "INBOX.CERT-TR"
+ # ensure each fetch would select its folder
+ with patch("advisory_rss.cert_tr.imap.connect_imap") as mock_connect:
+ m1 = _mock_imap([SAMPLE_RAW])
+ m2 = _mock_imap([SAMPLE_RAW])
+ mock_connect.side_effect = [m1, m2]
+ assert fetch_raw_emails(accs[0], ["siberguvenlik.gov.tr"], max_mails=1) is not None
+ assert fetch_raw_emails(accs[1], ["siberguvenlik.gov.tr"], max_mails=1) is not None
+ assert m1.select.call_args[0][0] == "INBOX"
+ assert m2.select.call_args[0][0] == "INBOX.CERT-TR"
+
+
+def test_gmail_host_allowed_and_proton_loopback():
+ from advisory_rss.cert_tr.imap import ALLOWED_HOSTS, connect_imap
+
+ assert "imap.gmail.com" in ALLOWED_HOSTS
+ assert "127.0.0.1" in ALLOWED_HOSTS
+
+ # Gmail host should be allowed (mock network to avoid real connection)
+ with patch("advisory_rss.cert_tr.imap.imaplib.IMAP4_SSL") as mock_ssl:
+ mock_inst = MagicMock()
+ mock_inst.login.side_effect = imaplib.IMAP4.error("mock auth fail")
+ mock_ssl.return_value = mock_inst
+ try:
+ connect_imap(
+ {
+ "host": "imap.gmail.com",
+ "port": "993",
+ "security": "SSL",
+ "email": "test@gmail.com",
+ "password": "bad",
+ }
+ )
+ except imaplib.IMAP4.error:
+ pass # expected login failure, host was allowed
+ except ValueError as ve:
+ raise AssertionError(f"imap.gmail.com should be allowed, got {ve}") from ve
+
+ # Evil host should be rejected before network
+ try:
+ connect_imap(
+ {
+ "host": "evil.com",
+ "port": "993",
+ "security": "SSL",
+ "email": "test@gmail.com",
+ "password": "bad",
+ }
+ )
+ raise AssertionError("evil host should be rejected")
+ except ValueError as ve:
+ assert "IMAP host" in str(ve)
diff --git a/tests/unit/test_cert_tr_parser.py b/tests/unit/test_cert_tr_parser.py
new file mode 100644
index 0000000..7d126dc
--- /dev/null
+++ b/tests/unit/test_cert_tr_parser.py
@@ -0,0 +1,57 @@
+from pathlib import Path
+
+from advisory_rss.cert_tr.parser import is_cert_tr_sender, parse_cert_tr_email
+
+
+def test_cert_tr_parse_sample():
+ raw = Path("tests/fixtures/cert_tr/CVE-2026-92083.eml").read_bytes()
+ adv = parse_cert_tr_email(raw)
+ assert adv is not None
+ assert adv.ghsa_id == "CERT-TR-CVE-2026-92083"
+ assert adv.cve_id == "CVE-2026-92083"
+ assert adv.source == "cert-tr"
+ assert adv.package_name == "pardus-etap-settings"
+ assert any(i["value"] == "CWE-862" for i in adv.identifiers)
+ assert "pardus-etap-settings" in adv.summary
+ assert adv.author_login == "cve@siberguvenlik.gov.tr"
+ assert adv.publisher_login == "siberguvenlik.gov.tr"
+ assert adv.repository_full_name == "cert-tr/pardus-etap-settings"
+ assert "https://nvd.nist.gov/vuln/detail/CVE-2026-92083" in adv.references
+
+
+def test_cert_tr_sender_allowlist():
+ assert is_cert_tr_sender(
+ "Ürün Güvenliği Koordinasyon Ekibi
", ["siberguvenlik.gov.tr"]
+ )
+ assert is_cert_tr_sender("cve@siberguvenlik.gov.tr", ["siberguvenlik.gov.tr"])
+ assert not is_cert_tr_sender("noreply@github.com", ["siberguvenlik.gov.tr"])
+ assert is_cert_tr_sender("test@usom.gov.tr", ["siberguvenlik.gov.tr", "usom.gov.tr"])
+
+
+def test_cert_tr_parser_second_mail_update():
+ raw1 = Path("tests/fixtures/cert_tr/CVE-2026-92083.eml").read_bytes()
+ adv1 = parse_cert_tr_email(raw1)
+ assert adv1 is not None
+ # Simulate update: same CVE but different body (e.g., patched)
+ raw2 = raw1.replace(b"CWE-862", b"CWE-79").replace(b"13:23:43 +0000", b"15:23:43 +0000")
+ adv2 = parse_cert_tr_email(raw2)
+ assert adv2 is not None
+ assert adv1.ghsa_id == adv2.ghsa_id
+ # updated time should differ? Date changed
+ assert adv2.identifiers != adv1.identifiers or adv2.updated_at != adv1.updated_at
+
+
+def test_cert_tr_parser_handles_quoted_printable_charset():
+ raw = Path("tests/fixtures/cert_tr/CVE-2026-92083.eml").read_bytes()
+ adv = parse_cert_tr_email(raw)
+ assert adv is not None
+ # Check Turkish chars decoded correctly
+ assert (
+ "TÜBİTAK" in adv.description or "TUBITAK" in adv.description or "BİLGEM" in adv.description
+ )
+
+
+def test_cert_tr_parser_no_cve_skipped():
+ raw = b"From: test@siberguvenlik.gov.tr\r\nSubject: Hello\r\nDate: Tue, 15 Sep 2026 13:23:43 +0000\r\nMessage-Id: \r\nContent-Type: text/plain; charset=utf-8\r\n\r\nJust a hello, no CVE here."
+ adv = parse_cert_tr_email(raw)
+ assert adv is None
diff --git a/tests/unit/test_cert_tr_rss.py b/tests/unit/test_cert_tr_rss.py
new file mode 100644
index 0000000..4aa9195
--- /dev/null
+++ b/tests/unit/test_cert_tr_rss.py
@@ -0,0 +1,68 @@
+from datetime import UTC, datetime
+
+from advisory_rss.cache.models import NormalizedAdvisory
+from advisory_rss.rss.builder import build_rss
+
+
+def test_mixed_rss():
+ gh = NormalizedAdvisory(
+ ghsa_id="GHSA-xxxx-yyyy-zzzz",
+ summary="GH vuln",
+ description="gh",
+ severity="high",
+ html_url="https://github.com/advisories/GHSA-xxxx",
+ author_login="user",
+ published_at=datetime(2026, 9, 20, tzinfo=UTC),
+ updated_at=datetime(2026, 9, 20, tzinfo=UTC),
+ source="github",
+ )
+ ct = NormalizedAdvisory(
+ ghsa_id="CERT-TR-CVE-2026-92083",
+ cve_id="CVE-2026-92083",
+ summary="pardus-etap-settings: CVE-2026-92083 (CWE-862)",
+ description="cert",
+ html_url="https://nvd.nist.gov/vuln/detail/CVE-2026-92083",
+ author_login="cve@siberguvenlik.gov.tr",
+ published_at=datetime(2026, 9, 15, tzinfo=UTC),
+ updated_at=datetime(2026, 9, 15, tzinfo=UTC),
+ source="cert-tr",
+ package_name="pardus-etap-settings",
+ identifiers=[{"type": "CWE", "value": "CWE-862"}],
+ )
+ xml = build_rss(
+ [gh, ct],
+ feed_title="T",
+ feed_link="https://example.com",
+ feed_description="d",
+ max_items=100,
+ ttl_minutes=10,
+ authenticated_user="user",
+ )
+ assert "source:github" in xml
+ assert "source:cert-tr" in xml
+ assert "[CERT-TR]" in xml
+ assert "cwe:CWE-862" in xml
+ # order: GHSA newer first if sorted inside builder
+ assert xml.index("GHSA-xxxx") < xml.index("CERT-TR-CVE")
+
+
+def test_cert_tr_rss_filter():
+ gh = NormalizedAdvisory(
+ ghsa_id="GHSA-1",
+ summary="g",
+ source="github",
+ updated_at=datetime(2026, 9, 20, tzinfo=UTC),
+ published_at=datetime(2026, 9, 20, tzinfo=UTC),
+ )
+ ct = NormalizedAdvisory(
+ ghsa_id="CERT-TR-CVE-1",
+ summary="c",
+ source="cert-tr",
+ updated_at=datetime(2026, 9, 15, tzinfo=UTC),
+ published_at=datetime(2026, 9, 15, tzinfo=UTC),
+ )
+ # simulate server filter
+ all_adv = [gh, ct]
+ filtered = [a for a in all_adv if getattr(a, "source", "github") == "cert-tr"]
+ assert len(filtered) == 1
+ assert filtered[0].ghsa_id == "CERT-TR-CVE-1"
diff --git a/tests/unit/test_cert_tr_settings.py b/tests/unit/test_cert_tr_settings.py
new file mode 100644
index 0000000..84b0052
--- /dev/null
+++ b/tests/unit/test_cert_tr_settings.py
@@ -0,0 +1,119 @@
+from advisory_rss.config.settings import Settings
+
+
+def test_single_account():
+ s = Settings(
+ _env_file="/dev/null",
+ ENABLE_CERT_TR="true",
+ PROTON_BRIDGE_EMAIL="a@proton.me",
+ PROTON_BRIDGE_PASSWORD="p1",
+ PROTON_IMAP_FOLDER="INBOX",
+ _env_file_encoding="utf-8",
+ )
+ accs = s.get_proton_accounts()
+ assert len(accs) == 1
+ assert accs[0]["email"] == "a@proton.me"
+ assert accs[0]["folder"] == "INBOX"
+ assert accs[0]["password"] == "p1"
+
+
+def test_multi_account_emails_folders():
+ s = Settings(
+ _env_file="/dev/null",
+ ENABLE_CERT_TR="true",
+ PROTON_BRIDGE_EMAILS="a@proton.me, b@proton.me",
+ PROTON_BRIDGE_PASSWORDS="p1,p2",
+ PROTON_IMAP_FOLDERS="INBOX,INBOX.CERT-TR",
+ _env_file_encoding="utf-8",
+ )
+ accs = s.get_proton_accounts()
+ assert len(accs) == 2
+ assert accs[0]["folder"] == "INBOX"
+ assert accs[1]["folder"] == "INBOX.CERT-TR"
+ assert accs[1]["password"] == "p2"
+
+
+def test_multi_account_fallback_single_password():
+ s = Settings(
+ _env_file="/dev/null",
+ ENABLE_CERT_TR="true",
+ PROTON_BRIDGE_EMAILS="a@proton.me, b@proton.me",
+ PROTON_BRIDGE_PASSWORD="single",
+ PROTON_IMAP_FOLDER="INBOX",
+ _env_file_encoding="utf-8",
+ )
+ accs = s.get_proton_accounts()
+ assert len(accs) == 2
+ assert accs[0]["password"] == "single"
+ assert accs[1]["password"] == "single"
+
+
+def test_folder_validation():
+ s = Settings(
+ _env_file="/dev/null", PROTON_IMAP_FOLDER="INBOX.CERT-TR", _env_file_encoding="utf-8"
+ )
+ assert s.proton_imap_folder == "INBOX.CERT-TR"
+
+
+def test_sender_allowlist():
+ s = Settings(
+ _env_file="/dev/null",
+ CERT_TR_SENDER_ALLOWLIST="siberguvenlik.gov.tr, usom.gov.tr",
+ _env_file_encoding="utf-8",
+ )
+ assert "siberguvenlik.gov.tr" in s.cert_tr_sender_list()
+ assert "usom.gov.tr" in s.cert_tr_sender_list()
+
+
+def test_gmail_single():
+ s = Settings(
+ _env_file="/dev/null",
+ ENABLE_CERT_TR="true",
+ GMAIL_EMAIL="a@gmail.com",
+ GMAIL_APP_PASSWORD="abcd efgh ijkl mnop",
+ GMAIL_IMAP_FOLDER="INBOX",
+ _env_file_encoding="utf-8",
+ )
+ accs = s.get_gmail_accounts()
+ assert len(accs) == 1
+ assert accs[0]["email"] == "a@gmail.com"
+ assert accs[0]["password"] == "abcd efgh ijkl mnop"
+ assert accs[0]["host"] == "imap.gmail.com"
+ assert accs[0]["security"] == "SSL"
+
+
+def test_gmail_multi_with_spaces():
+ s = Settings(
+ _env_file="/dev/null",
+ ENABLE_CERT_TR="true",
+ GMAIL_EMAILS="a@gmail.com,b@gmail.com",
+ GMAIL_APP_PASSWORDS="pass1,pass2 with spaces",
+ GMAIL_IMAP_FOLDERS="INBOX,INBOX.CERT",
+ _env_file_encoding="utf-8",
+ )
+ accs = s.get_gmail_accounts()
+ assert len(accs) == 2
+ assert accs[0]["password"] == "pass1"
+ assert accs[1]["password"] == "pass2 with spaces"
+ assert accs[1]["folder"] == "INBOX.CERT"
+
+
+def test_mixed_proton_gmail_unified():
+ s = Settings(
+ _env_file="/dev/null",
+ ENABLE_CERT_TR="true",
+ PROTON_BRIDGE_EMAIL="p@proton.me",
+ PROTON_BRIDGE_PASSWORD="pp",
+ GMAIL_EMAIL="g@gmail.com",
+ GMAIL_APP_PASSWORD="gp",
+ _env_file_encoding="utf-8",
+ )
+ accs = s.get_cert_tr_accounts()
+ assert len(accs) == 2
+ providers = {a["provider"] for a in accs}
+ assert providers == {"proton", "gmail"}
+ # proton should be loopback, gmail imap.gmail.com
+ proton = next(a for a in accs if a["provider"] == "proton")
+ gmail = next(a for a in accs if a["provider"] == "gmail")
+ assert proton["host"] == "127.0.0.1"
+ assert gmail["host"] == "imap.gmail.com"
diff --git a/tests/unit/test_normalize.py b/tests/unit/test_normalize.py
index 221e37e..0ec6ef4 100644
--- a/tests/unit/test_normalize.py
+++ b/tests/unit/test_normalize.py
@@ -74,7 +74,7 @@ def test_normalize_full():
assert adv.cve_id == raw.get("cve_id")
assert adv.severity == (raw.get("severity") or "medium").lower()
assert adv.summary == raw.get("summary")
- # author is per-user - compare to fixture, not hardcoded Lunixizm0
+
expected_author = (
(raw.get("author") or {}).get("login") if isinstance(raw.get("author"), dict) else None
)