"""Anmeldung an der Web-Oberflaeche mit den Benutzern von Proxmox VE. Geprueft wird ueber dieselbe Schnittstelle, die auch das Proxmox-Webinterface benutzt: POST /access/ticket auf der lokalen API. Passwoerter wandern also nirgends anders hin als ohnehin schon, und alle Realms (pam, pve, LDAP, AD ...) funktionieren automatisch mit. Wer sich anmelden darf, ist damit noch nicht entschieden: zusaetzlich muss der Benutzer auf der jeweiligen VM ein Recht besitzen (Vorgabe: VM.Snapshot). Sonst koennte jeder Benutzer mit einem gueltigen Kennwort saemtliche Dateien aller Gaeste lesen. """ from __future__ import annotations import json import logging import ssl import urllib.error import urllib.parse import urllib.request log = logging.getLogger("pvesnap.web.auth") API = "https://127.0.0.1:8006/api2/json" DEFAULT_PRIVILEGE = "VM.Snapshot" class AuthError(Exception): """Anmeldung fehlgeschlagen.""" def _unverified_context(): # Das Zertifikat von pveproxy ist selbstsigniert. Die Verbindung geht nur # an 127.0.0.1 und verlaesst den Host nicht. context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) context.check_hostname = False context.verify_mode = ssl.CERT_NONE return context def _pvesh_json(args): from ..proxmox import Proxmox, ProxmoxError try: return Proxmox()._json(args) except ProxmoxError as exc: raise AuthError(str(exc)) # --------------------------------------------------------------------------- # Realms # --------------------------------------------------------------------------- def list_realms(): """[(realm, Beschriftung)] - wie in der Auswahlliste des Proxmox-Logins.""" try: data = _pvesh_json(["get", "/access/domains"]) or [] except AuthError as exc: log.warning("Realms nicht lesbar: %s", exc) return [("pam", "Linux PAM standard authentication")] realms = [] for entry in data: realm = entry.get("realm") if not realm: continue comment = entry.get("comment") or entry.get("type") or realm realms.append((realm, "%s (%s)" % (realm, comment))) realms.sort(key=lambda item: (item[0] != "pam", item[0])) return realms or [("pam", "pam")] # --------------------------------------------------------------------------- # Anmeldung # --------------------------------------------------------------------------- def authenticate(username, realm, password): """Prueft die Zugangsdaten und gibt die vollstaendige Benutzerkennung zurueck.""" username = (username or "").strip() if not username or not password: raise AuthError("Benutzername und Passwort werden benoetigt.") if "@" in username: # "root@pam" im Feld eingetippt username, _, realm_from_name = username.partition("@") realm = realm_from_name or realm userid = "%s@%s" % (username, realm or "pam") data = urllib.parse.urlencode({"username": userid, "password": password}).encode() request = urllib.request.Request(API + "/access/ticket", data=data, method="POST") try: with urllib.request.urlopen(request, timeout=20, context=_unverified_context()) as response: payload = json.loads(response.read().decode("utf-8", "replace")) except urllib.error.HTTPError as exc: if exc.code == 401: raise AuthError("Benutzername oder Passwort falsch.") raise AuthError("Anmeldung fehlgeschlagen (HTTP %s)." % exc.code) except urllib.error.URLError as exc: raise AuthError("Der Proxmox-Dienst auf 127.0.0.1:8006 ist nicht " "erreichbar (%s)." % exc.reason) except (ValueError, OSError) as exc: raise AuthError("Anmeldung fehlgeschlagen: %s" % exc) result = (payload or {}).get("data") or {} if not result.get("ticket"): raise AuthError("Benutzername oder Passwort falsch.") if result.get("NeedTFA"): # Der zweite Faktor wird hier bewusst nicht abgefragt - lieber # abweisen als eine halbe Anmeldung durchwinken. raise AuthError("Fuer %s ist Zwei-Faktor-Authentifizierung aktiv. " "Diese Oberflaeche unterstuetzt das nicht - bitte einen " "Benutzer ohne zweiten Faktor verwenden." % userid) return result.get("username") or userid # --------------------------------------------------------------------------- # Rechte # --------------------------------------------------------------------------- def permissions(userid): """Rechtebaum eines Benutzers: {Pfad: {Recht: 1}}.""" try: data = _pvesh_json(["get", "/access/permissions", "--userid", userid]) except AuthError as exc: log.warning("Rechte von %s nicht lesbar: %s", userid, exc) return {} return data if isinstance(data, dict) else {} def may_access(userid, guest, privilege=DEFAULT_PRIVILEGE, tree=None, allowed_users=()): """Darf `userid` in die Dateien dieses Gastes sehen?""" if userid in (allowed_users or ()): return True if userid == "root@pam": return True # in Proxmox grundsaetzlich uneingeschraenkt tree = tree if tree is not None else permissions(userid) paths = ["/", "/vms", "/vms/%d" % guest.vmid] if getattr(guest, "pool", ""): paths.append("/pool/%s" % guest.pool) for path in paths: if (tree.get(path) or {}).get(privilege): return True return False