Deploymentcenter/tests/e2e/suite_hardware.py

#!/usr/bin/env python3
"""
Die Regel, welche Authenticatoren zugelassen sind.

WAS HIER GEPRUEFT WIRD - UND WARUM GERADE DAS

Der Anlass war ein Befund aus dem Betrieb: Ein Passwortverwalter liess sich
als zweiter Faktor registrieren, obwohl in den Registrierungsvorgaben
authenticatorAttachment = "cross-platform" stand. Die Angabe steuert nur, was
der Browser anbietet; sie ist keine Pruefung. Was traegt, ist die Attestation.

Der entscheidende Fall laesst sich ohne Hardware nachstellen: Ein
Authenticator, der "fmt": "none" schickt - also genau das, was ein
Passwortverwalter oder ein in das Betriebssystem eingebauter Passkey liefert -
muss unter der Regel abgewiesen werden und ohne sie durchkommen. Dazu braucht
es keine Kryptografie: Bei einer Registrierung mit "none" prueft der Server
keine Signatur, er liest den oeffentlichen Schluessel und legt ihn ab. Deshalb
kommt diese Reihe mit der Standardbibliothek aus.

WAS HIER NICHT GEPRUEFT WIRD

Der Gegenfall - ein echter Yubikey, der eine gueltige Kette vorlegt - laesst
sich von hier aus nicht erzeugen; dafuer braeuchte man Yubicos privaten
Wurzelschluessel, und genau darauf beruht die ganze Aussage. Ebenso fehlt der
Fall "packed mit selbst ausgestelltem Zertifikat", der belegt, dass ein x5c
allein nicht genuegt: Er verlangt eine echte Unterschrift ueber eine
Herausforderung, die erst zur Laufzeit feststeht, und damit eine
Krypto-Bibliothek. Beides laeuft in einem eigenen Werkzeug ausserhalb dieser
Suite.

Die Reihe raeumt am Ende auf und stellt die Regel auf "off" zurueck. Ohne das
liefe die naechste Reihe gegen eine Anlage, die Hardware verlangt - und
scheiterte an etwas, das sie gar nicht prueft.
"""

import base64
import http.cookiejar
import json
import re
import ssl
import urllib.error
import urllib.parse
import urllib.request

from harness import (test, erwarte, erwarte_status, ueberspringen, Fehlschlag)


# ---------------------------------------------------------------------------
# Ein oeffentlicher P-256-Punkt, fest eingebaut
# ---------------------------------------------------------------------------
# Warum fest und nicht erzeugt: Ein Punkt muss auf der Kurve liegen, sonst
# lehnt der Server schon beim Umwandeln nach PEM ab - und der Testfall
# scheiterte dann aus dem falschen Grund. Einen gueltigen Punkt zu WUERFELN
# verlangt dieselbe Bibliothek, die diese Suite nicht haben will. Also steht
# hier einer, der nachweislich auf P-256 liegt.
#
# Der zugehoerige private Schluessel ist bewusst nirgends abgelegt. Er wird
# nicht gebraucht: Bei einer Registrierung mit "none" unterschreibt niemand.
# Ein Schluessel, den man nicht braucht, ist ein Schluessel, der nicht
# verloren gehen kann.
PUBKEY_X = bytes.fromhex('6ee18ffb6fb32a97ac63039c118a9d507fb4624761df549a57b0f97457f85ca8')
PUBKEY_Y = bytes.fromhex('6291e3318ceadce692fd39d1822ffef9255f5eeb916b634694548c5d452b03cd')


# ---------------------------------------------------------------------------
# CBOR-Schreiber, gerade gross genug fuer ein Attestationsobjekt
# ---------------------------------------------------------------------------
def _kopf(major, wert):
    if wert < 24:
        return bytes([(major << 5) | wert])
    if wert < 0x100:
        return bytes([(major << 5) | 24, wert])
    if wert < 0x10000:
        return bytes([(major << 5) | 25]) + wert.to_bytes(2, 'big')
    return bytes([(major << 5) | 26]) + wert.to_bytes(4, 'big')


def _cbor(wert):
    if isinstance(wert, bytes):
        return _kopf(2, len(wert)) + wert
    if isinstance(wert, str):
        roh = wert.encode('utf-8')
        return _kopf(3, len(roh)) + roh
    if isinstance(wert, bool):
        return bytes([0xF5 if wert else 0xF4])
    if isinstance(wert, int):
        if wert >= 0:
            return _kopf(0, wert)
        return _kopf(1, -wert - 1)
    if isinstance(wert, list):
        return _kopf(4, len(wert)) + b''.join(_cbor(x) for x in wert)
    if isinstance(wert, dict):
        # Die Reihenfolge bleibt, wie sie eingetragen wurde. Der Server liest
        # die Abbildung als Ganzes; eine kanonische Sortierung braucht er
        # nicht, und sie nachzubilden waere Aufwand ohne Wirkung.
        out = _kopf(5, len(wert))
        for k, v in wert.items():
            out += _cbor(k) + _cbor(v)
        return out
    raise TypeError('CBOR: %r' % type(wert))


def _b64u(roh):
    return base64.urlsafe_b64encode(roh).decode('ascii').rstrip('=')


# ---------------------------------------------------------------------------
# Ein Sitzungsclient - die Suite selbst arbeitet mit Token, hier braucht es
# ein Cookie
# ---------------------------------------------------------------------------
class Sitzung:
    def __init__(self, cfg):
        self.basis = cfg['base_url'].rstrip('/')
        self.basic = cfg.get('basic_auth') or {}
        jar = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(jar),
            urllib.request.HTTPSHandler(context=ssl.create_default_context()),
        )
        self.csrf = None

    def _kopf(self, extra=None):
        k = {'User-Agent': 'dc-e2e/1.0', 'Accept': 'text/html,application/json'}
        if self.basic.get('user'):
            roh = ('%s:%s' % (self.basic['user'], self.basic['pass'])).encode()
            k['Authorization'] = 'Basic ' + base64.b64encode(roh).decode('ascii')
        k.update(extra or {})
        return k

    def hole(self, pfad):
        req = urllib.request.Request(self.basis + pfad, headers=self._kopf())
        try:
            with self.opener.open(req, timeout=45) as r:
                text = r.read().decode('utf-8', 'replace')
        except urllib.error.HTTPError as e:
            text = e.read().decode('utf-8', 'replace')
        treffer = re.search(r'name="csrf_token"\s+value="([^"]+)"', text)
        if treffer:
            self.csrf = treffer.group(1)
        return text

    def formular(self, pfad, felder):
        daten = dict(felder)
        if self.csrf:
            daten['csrf_token'] = self.csrf
        req = urllib.request.Request(
            self.basis + pfad,
            data=urllib.parse.urlencode(daten).encode(),
            headers=self._kopf({'Content-Type': 'application/x-www-form-urlencoded'}),
            method='POST')
        try:
            with self.opener.open(req, timeout=45) as r:
                return r.status, r.read().decode('utf-8', 'replace')
        except urllib.error.HTTPError as e:
            return e.code, e.read().decode('utf-8', 'replace')

    def json_post(self, pfad, rumpf):
        req = urllib.request.Request(
            self.basis + pfad,
            data=json.dumps(rumpf).encode(),
            headers=self._kopf({'Content-Type': 'application/json',
                                'X-CSRF-Token': self.csrf or ''}),
            method='POST')
        try:
            with self.opener.open(req, timeout=45) as r:
                return r.status, json.loads(r.read().decode('utf-8', 'replace'))
        except urllib.error.HTTPError as e:
            roh = e.read().decode('utf-8', 'replace')
            try:
                return e.code, json.loads(roh)
            except ValueError:
                return e.code, {'roh': roh[:400]}

    def kopfzeilen(self, pfad):
        """Antwortkopf einer Anfrage, Schluessel kleingeschrieben."""
        req = urllib.request.Request(self.basis + pfad, headers=self._kopf())
        try:
            with self.opener.open(req, timeout=45) as r:
                roh = r.headers
        except urllib.error.HTTPError as e:
            roh = e.headers
        return {k.lower(): v for k, v in roh.items()}

    def anmelden(self, benutzer, passwort):
        self.hole('/login.php')
        status, _ = self.formular('/login.php', {
            'username': benutzer,
            'password': passwort,
        })
        # Nach der Anmeldung steht das Token der Sitzung in index.php - das
        # aus dem Anmeldeformular gilt fuer die alte Sitzungskennung nicht
        # mehr.
        self.hole('/index.php')
        return status


def _schluessel_loeschen(ctx):
    """
    Raeumt die Hardwarekeys des Testkontos ab.

    Notwendig, weil diese Reihe selbst einen registriert - und ein Konto mit
    zweitem Faktor laesst sich mit dem Passwort allein nicht mehr anmelden.
    Ohne dieses Aufraeumen liefe der zweite Lauf gegen den Zwischenzustand
    und meldete "Anmeldung erforderlich", was nach einem Fehler der Anwendung
    aussieht und keiner ist.
    """
    benutzer = (ctx.cfg.get('admin') or {}).get('user') or 'admin'
    try:
        ctx.testctl('clear_credentials', username=benutzer)
    except Exception:
        # Aeltere Testinstanz ohne diese Aktion. Dann bleibt der Lauf auf das
        # Zuruecksetzen angewiesen - gemeldet wird es beim Anmelden.
        pass


def _sitzung(ctx):
    cfg = ctx.cfg
    if not (cfg.get('admin') or {}).get('user'):
        ueberspringen('In der Konfiguration steht kein Administratorkonto.')

    _schluessel_loeschen(ctx)

    s = Sitzung(cfg)
    s.anmelden(cfg['admin']['user'], cfg['admin']['pass'])

    if s.csrf is None:
        ueberspringen('Anmeldung am WebUI nicht moeglich - kein CSRF-Token erhalten.')

    return s


def _regel_setzen(s, stufe, aaguids=''):
    return s.formular('/index.php', {
        'action': 'save_webauthn_policy',
        '_return': 'sub-system-users',
        'policy_level': stufe,
        'policy_aaguids': aaguids,
    })


def _registrieren(s, bezeichnung):
    """Eine Registrierung mit fmt "none" - wie ein Passwortverwalter sie liefert."""
    status, vorgabe = s.json_post('/api/auth/v1/?action=register_options', {})
    if status != 200:
        raise Fehlschlag('register_options antwortete %d: %s' % (status, vorgabe))

    pk = vorgabe['publicKey']
    herausforderung = pk['challenge']
    rp_id = pk['rp']['id']

    import hashlib
    import os

    kennung = os.urandom(32)

    cose = {1: 2, 3: -7, -1: 1, -2: PUBKEY_X, -3: PUBKEY_Y}

    auth = hashlib.sha256(rp_id.encode()).digest()
    auth += bytes([0x41])                      # UP | AT
    auth += (0).to_bytes(4, 'big')             # Zaehler
    auth += b'\x00' * 16                       # aaguid: bei "none" immer null
    auth += len(kennung).to_bytes(2, 'big')
    auth += kennung
    auth += _cbor(cose)

    client = json.dumps({
        'type': 'webauthn.create',
        'challenge': herausforderung,
        'origin': s.basis,
        'crossOrigin': False,
    }).encode()

    attestation = _cbor({'fmt': 'none', 'attStmt': {}, 'authData': auth})

    return s.json_post('/api/auth/v1/?action=register_verify', {
        'clientDataJSON': _b64u(client),
        'attestationObject': _b64u(attestation),
        'label': bezeichnung,
    })


# ---------------------------------------------------------------------------
# Die Faelle
# ---------------------------------------------------------------------------

@test('hardware', 'Ohne Regel wird ein Software-Authenticator aufgenommen')
def t_ohne_regel(ctx):
    """
    Der Ausgangszustand. Er ist zugleich der Befund, um den es geht: Ein
    Authenticator ohne jeden Herkunftsnachweis kommt durch.
    """
    s = _sitzung(ctx)
    ctx['hw_sitzung'] = s

    status, _ = _regel_setzen(s, 'off')
    erwarte(status in (200, 302, 303), 'Regel liess sich nicht auf "off" stellen (%s).' % status)

    status, antwort = _registrieren(s, 'Testschluessel ohne Nachweis')

    erwarte(status == 200,
            'Ohne Regel muss die Aufnahme gelingen, war %d: %s' % (status, antwort))
    erwarte(antwort.get('attested') is False,
            'Ein Schluessel ohne Attestation darf nicht als Hardware gelten: %s' % antwort)


@test('hardware', 'Mit Regel wird derselbe Authenticator abgewiesen')
def t_mit_regel(ctx):
    """
    Derselbe Vorgang, dieselben Bytes - nur die Regel steht anders. Genau
    diese Gegenueberstellung ist die Aussage der Funktion.
    """
    s = ctx.get('hw_sitzung') or _sitzung(ctx)

    status, _ = _regel_setzen(s, 'register')
    erwarte(status in (200, 302, 303), 'Regel liess sich nicht auf "register" stellen (%s).' % status)

    status, antwort = _registrieren(s, 'Testschluessel unter Regel')

    erwarte(status == 403,
            'Unter der Regel muss die Aufnahme scheitern, war %d: %s' % (status, antwort))

    # Die Fehlerhuelle der API traegt den Code eine Ebene tiefer. Hier beide
    # Formen zu lesen ist keine Nachlaessigkeit: Der Testfall soll den
    # Zustand der Anwendung melden und nicht die Form ihrer Antwort.
    huelle = antwort.get('error')
    code = huelle.get('code') if isinstance(huelle, dict) else huelle
    erwarte(code == 'policy_rejected',
            'Erwartet wurde policy_rejected, kam: %s' % antwort)

    # Die Meldung muss den Grund nennen. Ein blosses "abgelehnt" liesse
    # offen, ob der Schluessel kaputt ist oder die Regel greift - und das ist
    # der Unterschied zwischen "anderen Schluessel holen" und "Einstellung
    # pruefen".
    text = (huelle.get('message') if isinstance(huelle, dict) else '') or ''
    erwarte('attestation' in text.lower() or 'hardware' in text.lower(),
            'Die Ablehnung nennt den Grund nicht: %s' % text)


@test('hardware', 'Die strenge Stufe sperrt sich nicht selbst aus')
def t_strict_bewacht(ctx):
    """
    Nach den beiden Faellen oben hat das Konto keinen attestierten Schluessel.
    "strict" wuerde die eigene Anmeldung sperren - und muss deshalb abgelehnt
    werden, statt sie zu sperren.
    """
    s = ctx.get('hw_sitzung') or _sitzung(ctx)

    status, _ = _regel_setzen(s, 'strict')
    erwarte(status in (200, 302, 303), 'Unerwarteter Status beim Umschalten (%s).' % status)

    # Die Ablehnung zeigt sich daran, dass die Stufe NICHT uebernommen wurde.
    # Das ist die Pruefung, auf die es ankommt: Eine Meldung ohne Wirkung
    # waere schlimmer als gar keine.
    seite = s.hole('/index.php')
    treffer = re.search(
        r'name="policy_level"\s+value="strict"[^>]*\schecked', seite)
    erwarte(treffer is None,
            'Die Stufe "strict" wurde uebernommen, obwohl kein attestierter '
            'Schluessel vorliegt - das sperrt den eigenen Zugang aus.')


@test('hardware', 'Eine unbrauchbare Modellliste wird abgewiesen')
def t_allowlist_pruefung(ctx):
    """
    Eine Geraetekennung, die keine ist, darf nicht stillschweigend als leere
    Liste durchgehen - das hiesse "alles erlaubt", wo jemand "nur dieses
    Modell" gemeint hat.
    """
    s = ctx.get('hw_sitzung') or _sitzung(ctx)

    status, _ = _regel_setzen(s, 'register', aaguids='kein-gueltiges-geraet')
    erwarte(status in (200, 302, 303), 'Unerwarteter Status (%s).' % status)

    seite = s.hole('/index.php')
    erwarte('kein-gueltiges-geraet' not in seite,
            'Die unbrauchbare Angabe wurde gespeichert.')


@test('hardware', 'Aufraeumen: Regel steht wieder auf "off"')
def t_aufraeumen(ctx):
    """
    Kein Test im eigentlichen Sinn, sondern die Rueckgabe eines geliehenen
    Zustands. Bliebe die Regel stehen, liefe die naechste Reihe gegen eine
    Anlage, die Hardware verlangt.
    """
    s = ctx.get('hw_sitzung') or _sitzung(ctx)

    status, _ = _regel_setzen(s, 'off')
    erwarte(status in (200, 302, 303), 'Regel liess sich nicht zuruecksetzen (%s).' % status)

    seite = s.hole('/index.php')
    treffer = re.search(r'name="policy_level"\s+value="off"[^>]*\schecked', seite)
    erwarte(treffer is not None, 'Die Regel steht nicht wieder auf "off".')

    # Der in dieser Reihe angelegte Schluessel muss ebenfalls weg. Bliebe er
    # stehen, verlangte das Testkonto beim naechsten Lauf einen zweiten
    # Faktor, den niemand vorzeigen kann.
    _schluessel_loeschen(ctx)


# ---------------------------------------------------------------------------
# Haertung der Anmeldewege
#
# Ergebnis einer Sicherheitsbegutachtung. Der rote Faden aller Befunde: Die
# WebAuthn-Zeremonie war in Ordnung, der zweite Faktor trotzdem umgehbar -
# ueber Wege DANEBEN. Was hier geprueft wird, sind genau diese Wege.
# ---------------------------------------------------------------------------

@test('haertung', 'Die Anmeldung verraet nicht, ob es ein Konto gibt')
def t_keine_enumeration(ctx):
    """
    Gesperrt und gezaehlt wird nur, was existiert. Unterscheidet die Seite
    sichtbar zwischen "Konto gesperrt" und "ungueltige Daten", genuegen
    fuenf Versuche pro Name, um seine Existenz zu bestaetigen - und die
    sorgfaeltige Laufzeitangleichung ueber dummyHash() waere umsonst.

    Ein Fehlversuch gegen das echte Konto ist dabei unvermeidlich; die
    Kontosperre greift erst beim fuenften.
    """
    s = Sitzung(ctx.cfg)

    s.hole('/login.php')
    _, unbekannt = s.formular('/login.php', {
        'username': 'gibt-es-garantiert-nicht-4711', 'password': 'beliebig'})

    s2 = Sitzung(ctx.cfg)
    s2.hole('/login.php')
    _, vorhanden = s2.formular('/login.php', {
        'username': (ctx.cfg.get('admin') or {}).get('user') or 'admin',
        'password': 'garantiert-falsch'})

    def meldung(seite):
        treffer = re.search(r'<div class="alert-danger">(.*?)</div>', seite, re.S)
        return re.sub(r'\s+', ' ', treffer.group(1)).strip() if treffer else ''

    a, b = meldung(unbekannt), meldung(vorhanden)

    erwarte(a != '', 'Kein Fehlertext fuer den unbekannten Namen gefunden.')
    erwarte(a == b,
            'Die Meldungen unterscheiden sich und verraten damit die Existenz '
            'des Kontos:\n  unbekannt: %r\n  vorhanden: %r' % (a, b))


@test('haertung', 'Die Sicherheits-Header sind gesetzt')
def t_header(ctx):
    """
    HSTS und CSP. Zur CSP gehoert die Einschraenkung, dass sie wegen der
    Inline-Skripte in index.php 'unsafe-inline' enthaelt und deshalb NICHT
    gegen XSS traegt - geprueft wird hier, dass die Richtlinien ueberhaupt
    ankommen, nicht dass sie alles loesen.
    """
    s = Sitzung(ctx.cfg)
    kopf = s.kopfzeilen('/login.php')

    hsts = kopf.get('strict-transport-security', '')
    erwarte('max-age=' in hsts, 'Kein Strict-Transport-Security: %r' % hsts)

    csp = kopf.get('content-security-policy', '')
    for pflicht in ("frame-ancestors 'none'", "base-uri 'none'", "form-action 'self'"):
        erwarte(pflicht in csp, 'In der CSP fehlt %r: %r' % (pflicht, csp))


@test('haertung', 'Die Gesundheitsauskunft nennt den WebAuthn-Anker')
def t_anker(ctx):
    """
    rpId und expectedOrigin stammen nicht mehr aus dem Host-Header. Die
    Folge ist, dass eine Anlage ohne app.url den Hardwarekey abweist - und
    genau das soll man sehen koennen, BEVOR jemand sich anzumelden versucht.
    """
    r = ctx.client.get('/api/health', token=None)
    erwarte_status(r, 200)

    anker = r.feld('checks', 'webauthn_anchor')
    erwarte(isinstance(anker, dict),
            'Die Gesundheitsauskunft nennt keinen webauthn_anchor: %s' % r.kurz(300))
    erwarte(anker.get('ok') is True,
            'Der WebAuthn-Anker ist nicht gesetzt - app.url fehlt: %s' % anker)