# SPDX-License-Identifier: MIT """Administrator authentication; never an alternative to broker authentication. The gateway is external infrastructure. No offline password fallback, test mode, HTTP fallback, fabricated token or automatically approved login exists here. """ import asyncio import contextlib import getpass import hashlib import json import platform import resource import sys import threading import time from urllib.parse import urlsplit import aiohttp from yarl import URL class AuthenticationFailed(Exception): def __init__(self): super().__init__('AUTHENTICATION_FAILED') class AdministratorGate: PROMPT = 'AlphaBot gateway password: ' TTL = 3600 def __init__(self, settings, environment, on_failure=None): self.settings = settings self.on_failure = on_failure self.base = environment.get('LIVE_AUTH_GATEWAY_URL', '') if self.base and '://' not in self.base: self.base = 'https://' + self.base try: parsed = urlsplit(self.base) if (parsed.scheme != 'https' or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ('', '/') or parsed.hostname != settings.get('gateway_hostname') or (parsed.port or 443) != settings.get('gateway_port', 443)): raise ValueError except ValueError: raise AuthenticationFailed() from None self.hostname = parsed.hostname self.base = self.base.rstrip('/') self.headers = {} cf_id, cf_secret = environment.get('CF_ACCESS_CLIENT_ID'), environment.get('CF_ACCESS_CLIENT_SECRET') if bool(cf_id) != bool(cf_secret): raise AuthenticationFailed() if cf_id: self.headers = {'CF-Access-Client-Id': cf_id, 'CF-Access-Client-Secret': cf_secret} self.revoked = threading.Event() self.revoked.set() self.closed = threading.Event() self.deadline = 0 self.session = None self.bound_account = None self.generation = 0 self.mutex = threading.RLock() self.monitor = None def _prompt(self): if (platform.system() != 'Darwin' or not sys.stdin.isatty() or threading.current_thread() is not threading.main_thread()): raise AuthenticationFailed() resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) # Hidden input MUST remain on the main thread; a background worker never prompts. return getpass.getpass(self.PROMPT) async def open(self): self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15), trust_env=False, cookie_jar=aiohttp.CookieJar()) try: await self.authenticate() except BaseException: await self.close() raise self.monitor = threading.Thread(target=self._watch, name='administrator-expiry', daemon=True) self.monitor.start() return self def _watch(self): while not self.closed.wait(0.1): with self.mutex: if time.monotonic() >= self.deadline: self.revoked.set() def require(self): with self.mutex: if self.closed.is_set() or self.revoked.is_set() or time.monotonic() >= self.deadline: self.revoked.set() raise AuthenticationFailed() @property def expires_in(self): with self.mutex: return max(0, int(self.deadline - time.monotonic())) def _cookie_valid(self): usable = self.session.cookie_jar.filter_cookies(URL(self.base)) names = set(usable) return any(cookie.key in names and cookie['secure'] and cookie['httponly'] and cookie['path'] == '/' and cookie['domain'].lstrip('.') == self.hostname for cookie in self.session.cookie_jar) async def _post(self, path, body, authenticated=True): if authenticated: self.require() if not self._cookie_valid(): raise AuthenticationFailed() async with self.session.post(self.base + path, json=body, headers=self.headers, allow_redirects=False) as response: if response.status != 200: raise AuthenticationFailed() raw = await response.content.read(65537) if len(raw) > 65536: raise AuthenticationFailed() result = json.loads(raw) if not isinstance(result, dict): raise AuthenticationFailed() return result async def authenticate(self): self.revoked.set() self.session.cookie_jar.clear() password = challenge = None try: password = self._prompt() challenge = hashlib.sha256(password.encode('utf-8')).hexdigest() password = None started = time.monotonic() result = await self._post('/api/v1/auth/gate', {'gate_challenge': challenge}, authenticated=False) if (result.get('status') != 'AUTHENTICATED' or type(result.get('session_token_expires_sec')) is not int or result['session_token_expires_sec'] != self.TTL or not self._cookie_valid()): raise AuthenticationFailed() with self.mutex: self.deadline = started + self.TTL if self.deadline <= time.monotonic(): raise AuthenticationFailed() self.generation += 1 self.revoked.clear() except BaseException: self.revoked.set() self.session.cookie_jar.clear() if self.on_failure: self.on_failure('AUTHENTICATION_FAILED') raise AuthenticationFailed() from None finally: password = challenge = None # Immutable Python strings cannot be securely wiped. async def bootstrap(self, broker, account_id, key, secret): try: if not account_id or not key or not secret: raise AuthenticationFailed() result = await self._post('/v1/broker-session', {'broker': broker, 'account_id': account_id, 'api_key': key, 'secret_key': secret}) if result.get('bound') is not True or result.get('account_id') != account_id: raise AuthenticationFailed() self.bound_account = account_id except BaseException: self.revoked.set() if self.on_failure: self.on_failure('AUTHENTICATION_FAILED') raise AuthenticationFailed() from None async def renew_if_due(self): # Renewal is performed BEFORE strategy invocation after expiry; no silent refresh. if self.revoked.is_set() or time.monotonic() >= self.deadline: await self.authenticate() self.require() async def close(self): self.closed.set() self.revoked.set() if self.session: self.session.cookie_jar.clear() await self.session.close() if self.monitor: self.monitor.join(timeout=0.3) self.headers.clear()