"""Private cookie/CSRF client for IKAR deployment checks. Never logs credentials.""" import json from http.cookies import SimpleCookie from urllib.parse import quote from urllib.request import Request, urlopen class AdminHttp: def __init__(self, base='http://127.0.0.1:5099'): self.base = base.rstrip('/') self.cookies = {} self.csrf = None def request(self, path, method='GET', payload=None, reason=None): headers = {'Accept': 'application/json', 'X-Forwarded-Proto': 'https'} if self.cookies: headers['Cookie'] = '; '.join(key + '=' + value for key, value in self.cookies.items()) if method not in ('GET', 'HEAD') and self.csrf: headers['X-Ikar-CSRF'] = self.csrf if reason: headers['X-Admin-Reason'] = quote(reason) data = None if payload is None else json.dumps(payload).encode() if data is not None: headers['Content-Type'] = 'application/json' with urlopen(Request(self.base + path, data=data, method=method, headers=headers), timeout=30) as response: for value in response.headers.get_all('Set-Cookie', []): cookie = SimpleCookie(); cookie.load(value) for key, morsel in cookie.items(): if morsel['max-age'] == '0' or not morsel.value: self.cookies.pop(key, None) else: self.cookies[key] = morsel.value body = response.read() return response.status, json.loads(body) if body else None def login(self, settings): _, state = self.request('/admin/api/session'); self.csrf = state['csrfToken'] status, _ = self.request('/admin/api/session/login', 'POST', {'username': settings['Admin__Username'], 'password': settings['Admin__Password']}) if status != 204: raise RuntimeError('Admin login failed') _, state = self.request('/admin/api/session'); self.csrf = state['csrfToken'] if not state['authenticated']: raise RuntimeError('Admin cookie authentication failed') def logout(self): self.request('/admin/api/session/logout', 'POST')