147 lines
4.5 KiB
Python
147 lines
4.5 KiB
Python
import logging
|
|
import os
|
|
import threading
|
|
import urllib.error
|
|
import urllib.request
|
|
from urllib.parse import urlparse
|
|
|
|
from app.models import ICON_DIR, SessionLocal, Dapp, Token
|
|
|
|
logger = logging.getLogger("idrs.verify")
|
|
|
|
TOKENS_ENABLED = os.environ.get("IDRS_VERIFY_TOKENS", "").strip().lower() in ("1", "true", "yes", "on")
|
|
DAPPS_ENABLED = os.environ.get("IDRS_VERIFY_DAPPS", "1").strip().lower() not in ("0", "false", "off", "no")
|
|
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, accept: str = "image/*,*/*") -> bool:
|
|
req = urllib.request.Request(
|
|
url, headers={"User-Agent": USER_AGENT, "Accept": accept}
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
|
|
return 200 <= resp.status < 300
|
|
except urllib.error.HTTPError as e:
|
|
logger.warning("http %s for %s", e.code, url)
|
|
return False
|
|
except Exception as e:
|
|
logger.warning("network error %r for %s", e, url)
|
|
return False
|
|
|
|
|
|
class Verifier(threading.Thread):
|
|
def __init__(self, name: str, enabled: bool):
|
|
super().__init__(daemon=True, name=name)
|
|
self._enabled = enabled
|
|
self._stop = threading.Event()
|
|
self._fail_counts = {}
|
|
|
|
def stop(self):
|
|
self._stop.set()
|
|
|
|
def start(self):
|
|
if not self._enabled:
|
|
logger.info("%s disabled", self.name)
|
|
return
|
|
super().start()
|
|
|
|
def run(self):
|
|
while not self._stop.is_set():
|
|
try:
|
|
self._check_once()
|
|
except Exception:
|
|
logger.exception("%s pass failed", self.name)
|
|
self._stop.wait(INTERVAL)
|
|
|
|
def _check_once(self):
|
|
session = SessionLocal()
|
|
try:
|
|
for item in self._items(session):
|
|
if self._skip(item):
|
|
continue
|
|
ok = self._reachable(item)
|
|
key = self._key(item)
|
|
if ok:
|
|
self._fail_counts[key] = 0
|
|
live = True
|
|
else:
|
|
fails = self._fail_counts.get(key, 0) + 1
|
|
self._fail_counts[key] = fails
|
|
live = fails < GRACE_PASSES and bool(item.is_live)
|
|
if bool(item.is_live) != live:
|
|
item.is_live = live
|
|
logger.info("%s is_live -> %s (%s)", key, live, self._label(item))
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
def _items(self, session):
|
|
raise NotImplementedError
|
|
|
|
def _key(self, item):
|
|
raise NotImplementedError
|
|
|
|
def _label(self, item):
|
|
raise NotImplementedError
|
|
|
|
def _reachable(self, item) -> bool:
|
|
raise NotImplementedError
|
|
|
|
def _skip(self, item) -> bool:
|
|
return False
|
|
|
|
|
|
class TokenVerifier(Verifier):
|
|
def __init__(self):
|
|
super().__init__(name="idrs-verify-tokens", enabled=TOKENS_ENABLED)
|
|
|
|
def _items(self, session):
|
|
return session.query(Token).all()
|
|
|
|
def _key(self, item):
|
|
return item.idfToken
|
|
|
|
def _label(self, item):
|
|
return item.iconUrl or ""
|
|
|
|
def _reachable(self, item) -> bool:
|
|
url = (item.iconUrl or "").strip()
|
|
if not url:
|
|
return False
|
|
if urlparse(url).hostname == PUBLIC_HOST:
|
|
return _local_reachable(item.filename or "")
|
|
return _http_reachable(url, "image/*,*/*")
|
|
|
|
|
|
class DappVerifier(Verifier):
|
|
def __init__(self):
|
|
super().__init__(name="idrs-verify-dapps", enabled=DAPPS_ENABLED)
|
|
|
|
def _items(self, session):
|
|
return session.query(Dapp).all()
|
|
|
|
def _key(self, item):
|
|
return item.id
|
|
|
|
def _label(self, item):
|
|
return item.link or ""
|
|
|
|
def _reachable(self, item) -> bool:
|
|
url = (item.link or "").strip()
|
|
if not url:
|
|
return False
|
|
return _http_reachable(url, "text/html,application/xhtml+xml,*/*")
|
|
|
|
def _skip(self, item) -> bool:
|
|
return bool(getattr(item, "is_live_override", False)) |