proxmox-snapshot-service/pvesnap/web/auth.py

"""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