91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
from app.models import ICON_DIR, SessionLocal, Token
|
|
|
|
logger = logging.getLogger("idrs.verify")
|
|
|
|
ENABLED = os.environ.get("IDRS_VERIFY_ENABLED", "").strip().lower() in ("1", "true", "yes", "on")
|
|
PUBLIC_BASE_URL = os.environ.get("IDRS_PUBLIC_URL", "https://idrs.databisnis.id").rstrip("/")
|
|
PUBLIC_HOST = urlparse(PUBLIC_BASE_URL).hostname or "idrs.databisnis.id"
|
|
INTERVAL = int(os.environ.get("IDRS_VERIFY_INTERVAL", "600"))
|
|
GRACE_PASSES = 2
|
|
HTTP_TIMEOUT = 3
|
|
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36"
|
|
|
|
|
|
def _local_reachable(filename: str) -> bool:
|
|
token_dir = ICON_DIR / "token"
|
|
path = (token_dir / filename).resolve()
|
|
return path.is_file() and token_dir.resolve() in path.parents
|
|
|
|
|
|
def _http_reachable(url: str) -> bool:
|
|
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "image/*,*/*"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
|
|
return 200 <= resp.status < 300
|
|
except urllib.error.HTTPError as e:
|
|
logger.warning("icon http %s for %s", e.code, url)
|
|
return False
|
|
except Exception as e:
|
|
logger.warning("icon network error %r for %s", e, url)
|
|
return False
|
|
|
|
|
|
def reachable(token) -> bool:
|
|
url = (token.iconUrl or "").strip()
|
|
if not url:
|
|
return False
|
|
if urlparse(url).hostname == PUBLIC_HOST:
|
|
return _local_reachable(token.filename or "")
|
|
return _http_reachable(url)
|
|
|
|
|
|
class Verifier(threading.Thread):
|
|
def __init__(self):
|
|
super().__init__(daemon=True, name="idrs-verify")
|
|
self._stop = threading.Event()
|
|
self._fail_counts = {}
|
|
|
|
def stop(self):
|
|
self._stop.set()
|
|
|
|
def start(self):
|
|
if not ENABLED:
|
|
logger.info("verifier disabled (IDRS_VERIFY_ENABLED not set)")
|
|
return
|
|
super().start()
|
|
|
|
def run(self):
|
|
while not self._stop.is_set():
|
|
try:
|
|
self._check_once()
|
|
except Exception:
|
|
logger.exception("verify pass failed")
|
|
self._stop.wait(INTERVAL)
|
|
|
|
def _check_once(self):
|
|
session = SessionLocal()
|
|
try:
|
|
for token in session.query(Token).all():
|
|
ok = reachable(token)
|
|
if ok:
|
|
self._fail_counts[token.idfToken] = 0
|
|
live = True
|
|
else:
|
|
fails = self._fail_counts.get(token.idfToken, 0) + 1
|
|
self._fail_counts[token.idfToken] = fails
|
|
live = fails < GRACE_PASSES
|
|
if bool(token.is_live) != live:
|
|
token.is_live = live
|
|
logger.info("token %s is_live -> %s (%s)", token.idfToken, live, token.iconUrl)
|
|
session.commit()
|
|
finally:
|
|
session.close() |