Skey/tests/test_presence.py

"""The touch button: requests wait for a press, with keepalives and cancel."""

import os
import struct
import threading
import time

import pytest
from fido2.ctap import STATUS, CtapError
from fido2.ctap1 import APDU, ApduError, Ctap1
from fido2.ctap2 import Ctap2
from fido2.hid import CTAPHID
from fido2 import cbor

RP = {"id": "example.com", "name": "Example"}
ALICE = {"id": b"alice-id", "name": "alice"}
ES256 = [{"type": "public-key", "alg": -7}]
TOUCH_DELAY = 0.3


def make(ctap, **kwargs):
    return ctap.make_credential(os.urandom(32), RP, ALICE, ES256, **kwargs)


def test_waits_for_touch_with_keepalives(button_sim):
    ctap = Ctap2(button_sim.device)
    statuses = []
    button_sim.press_after(TOUCH_DELAY)
    start = time.monotonic()

    att = make(ctap, on_keepalive=statuses.append)

    assert time.monotonic() - start >= TOUCH_DELAY
    assert STATUS.UPNEEDED in statuses
    assert att.auth_data.is_user_present()


def test_assertion_waits_for_touch(button_sim):
    ctap = Ctap2(button_sim.device)
    button_sim.press_after(0.05)  # presses only count once the request has arrived
    cred = make(ctap).auth_data.credential_data
    allow = [{"type": "public-key", "id": cred.credential_id}]

    button_sim.press_after(TOUCH_DELAY)
    start = time.monotonic()
    assertion = ctap.get_assertion(RP["id"], os.urandom(32), allow_list=allow)

    assert time.monotonic() - start >= TOUCH_DELAY
    assert assertion.auth_data.is_user_present()


def test_press_before_request_does_not_count(button_sim):
    ctap = Ctap2(button_sim.device)
    button_sim.press()
    time.sleep(0.1)
    button_sim.press_after(TOUCH_DELAY)
    start = time.monotonic()

    make(ctap)

    assert time.monotonic() - start >= TOUCH_DELAY


def test_silent_assertion_needs_no_touch(button_sim):
    ctap = Ctap2(button_sim.device)
    button_sim.press_after(0.05)  # presses only count once the request has arrived
    cred = make(ctap).auth_data.credential_data
    allow = [{"type": "public-key", "id": cred.credential_id}]
    assertion = ctap.get_assertion(RP["id"], os.urandom(32), allow_list=allow, options={"up": False})
    assert not assertion.auth_data.is_user_present()


def test_cancel(button_sim):
    ctap = Ctap2(button_sim.device)
    cancel = threading.Event()
    threading.Timer(TOUCH_DELAY, cancel.set).start()

    with pytest.raises(CtapError) as e:
        make(ctap, event=cancel)

    assert e.value.code == CtapError.ERR.KEEPALIVE_CANCEL
    assert ctap.get_info().versions  # still responsive


def test_u2f_fails_until_touched(button_sim):
    ctap1 = Ctap1(button_sim.device)
    app_id, challenge = os.urandom(32), os.urandom(32)
    with pytest.raises(ApduError) as e:
        ctap1.register(challenge, app_id)
    assert e.value.code == APDU.USE_NOT_SATISFIED

    # Hosts retry; a press since the last attempt lets the next one through.
    button_sim.press()
    reg = ctap1.register(challenge, app_id)
    reg.verify(app_id, challenge)

    with pytest.raises(ApduError) as e:
        ctap1.authenticate(challenge, app_id, reg.key_handle)
    assert e.value.code == APDU.USE_NOT_SATISFIED
    button_sim.press()
    ctap1.authenticate(challenge, app_id, reg.key_handle).verify(app_id, challenge, reg.public_key)


def send_message(conn, cid, cmd, data):
    conn.write_packet((struct.pack(">IBH", cid, 0x80 | cmd, len(data)) + data[:57]).ljust(64, b"\0"))
    data, seq = data[57:], 0
    while data:
        conn.write_packet((struct.pack(">IB", cid, seq) + data[:59]).ljust(64, b"\0"))
        data, seq = data[59:], seq + 1


def read_until(conn, cid, cmd):
    """Reads packets until an init packet for (cid, cmd); returns all seen."""
    seen = []
    while True:
        packet = conn.read_packet()
        r_cid, r_cmd = struct.unpack_from(">IB", packet)
        seen.append((r_cid, r_cmd, packet))
        if (r_cid, r_cmd) == (cid, 0x80 | cmd):
            return seen


def test_other_channels_busy_while_waiting(button_sim):
    conn = button_sim.device._connection
    cid = button_sim.device._channel_id
    request = bytes([Ctap2.CMD.MAKE_CREDENTIAL]) + cbor.encode({1: os.urandom(32), 2: RP, 3: ALICE, 4: ES256})
    send_message(conn, cid, CTAPHID.CBOR, request)
    read_until(conn, cid, CTAPHID.KEEPALIVE)

    send_message(conn, cid + 1, CTAPHID.PING, b"hi")
    seen = read_until(conn, cid + 1, CTAPHID.ERROR)
    assert seen[-1][2][7] == CtapError.ERR.CHANNEL_BUSY

    button_sim.press()
    seen = read_until(conn, cid, CTAPHID.CBOR)
    assert seen[-1][2][7] == 0  # CTAP2_OK
    assert all(r_cid == cid for r_cid, _, _ in seen)
    # Drain the rest of the response so the connection closes cleanly.
    length = struct.unpack_from(">H", seen[-1][2], 5)[0]
    for _ in range((max(0, length - 57) + 58) // 59):
        conn.read_packet()