authentik.events.context_processors.mmdb

 1"""Common logic for reading MMDB files"""
 2
 3from pathlib import Path
 4
 5from geoip2.database import Reader
 6from structlog.stdlib import get_logger
 7
 8from authentik.events.context_processors.base import EventContextProcessor
 9
10
11class MMDBContextProcessor(EventContextProcessor):
12    """Common logic for reading MaxMind DB files, including re-loading if the file has changed"""
13
14    def __init__(self):
15        self.reader: Reader | None = None
16        self._last_mtime: float = 0.0
17        self.logger = get_logger()
18        self.load()
19
20    def path(self) -> str | None:
21        """Get the path to the MMDB file to load"""
22        raise NotImplementedError
23
24    def load(self):
25        """Get GeoIP Reader, if configured, otherwise none"""
26        path = self.path()
27        if path == "" or not path:
28            return
29        try:
30            self.reader = Reader(path)
31            self._last_mtime = Path(path).stat().st_mtime
32            self.logger.info("Loaded MMDB database", last_write=self._last_mtime, file=path)
33        except OSError as exc:
34            self.logger.warning("Failed to load MMDB database", path=path, exc=exc)
35
36    def check_expired(self):
37        """Check if the modification date of the MMDB database has
38        changed, and reload it if so"""
39        path = self.path()
40        if path == "" or not path:
41            return
42        try:
43            mtime = Path(path).stat().st_mtime
44            diff = self._last_mtime < mtime
45            if diff > 0:
46                self.logger.info("Found new MMDB Database, reopening", diff=diff, path=path)
47                self.load()
48        except OSError as exc:
49            self.logger.warning("Failed to check MMDB age", exc=exc)
50
51    def configured(self) -> bool:
52        """Return true if this context processor is configured"""
53        return bool(self.reader)