devplacepy/tests/unit/services/community/login.py

# retoor <retoor@molodetz.nl>

import base64

import pytest

from devplacepy.services.community import runtime
from devplacepy.services.community.discoverer import LoginLearner
from devplacepy.services.community.extract import detect_captcha, find_otp_form, parse_html, totp
from devplacepy.services.community.runtime import (
    CaptchaRequired,
    Collector,
    LoginFailed,
    SecondFactorRequired,
    code_challenge,
)
from tests.conftest import run_async
from tests.unit.services.community.fakes import FakeFetcher


async def no_sleep(_seconds):
    return None


RFC_SECRET = base64.b32encode(b"12345678901234567890").decode()


@pytest.mark.parametrize(
    "moment,expected",
    [(59, "94287082"), (1111111109, "07081804"), (1111111111, "14050471"), (1234567890, "89005924"), (2000000000, "69279037"), (20000000000, "65353130")],
)
def test_totp_matches_the_rfc_6238_vectors(moment, expected):
    assert totp(RFC_SECRET, moment, digits=8) == expected


def test_totp_default_is_six_digits_zero_padded_and_tolerates_formatting():
    code = totp(RFC_SECRET.lower(), 59)
    assert len(code) == 6 and code.isdigit()
    spaced = " ".join(RFC_SECRET[index : index + 4] for index in range(0, len(RFC_SECRET), 4))
    assert totp(spaced, 59) == code
    assert totp(RFC_SECRET.rstrip("="), 59) == code
    assert totp(RFC_SECRET, 59 + 30) != code
    assert totp(RFC_SECRET, 59) == totp(RFC_SECRET, 30)


OTP_PAGE = """<html><body><form action="/2fa" method="post">
<input type="hidden" name="csrf" value="t2"><input type="text" name="otp_code" autocomplete="one-time-code"><button>Verify</button></form></body></html>"""
LOGIN_PAGE = """<html><body><form action="/login" method="post">
<input type="hidden" name="csrf" value="t1"><input type="text" name="user"><input type="password" name="pass"><button>Sign in</button></form></body></html>"""
HOME_OPEN = "<html><body><p>Welcome back to the forum, you are signed in</p></body></html>"


def test_otp_forms_and_captcha_markers_are_recognised():
    form = find_otp_form(parse_html(OTP_PAGE), "https://x.test/login")
    assert form == {"action": "https://x.test/2fa", "method": "POST", "field": "otp_code", "fields": {"csrf": "t2"}}
    assert find_otp_form(parse_html(LOGIN_PAGE), "https://x.test/") is None
    named = parse_html('<form><input type="text" name="verification_code"></form>')
    assert find_otp_form(named, "https://x.test/")["field"] == "verification_code"
    hidden_token = parse_html('<form><input type="hidden" name="token"><input type="text" name="user"></form>')
    assert find_otp_form(hidden_token, "https://x.test/") is None
    assert detect_captcha('<div class="g-recaptcha" data-sitekey="x"></div>') and detect_captcha("<div class='cf-turnstile'></div>")
    assert not detect_captcha("<p>no puzzle here</p>")


def form_schema():
    return {
        "community": "club",
        "origin": "club.test",
        "base_url": "https://club.test/",
        "feeds": [],
        "limits": {"delay_seconds": 0, "retries": 1},
        "login": {
            "mode": "form", "url": "https://club.test/login", "action": "https://club.test/login", "method": "POST",
            "fields": {"csrf": "t1"}, "username_field": "user", "password_field": "pass", "identity": "username",
        },
    }


def club(state, challenge=True, good_code="123456"):
    def login(data):
        if data.get("user") == "bob" and data.get("pass") == "pw":
            if challenge:
                return (200, OTP_PAGE)
            state["in"] = True
            return (200, "ok")
        return (200, LOGIN_PAGE)

    def second(data):
        state["codes"] = state.get("codes", []) + [data.get("otp_code")]
        state["in"] = data.get("otp_code") == good_code and data.get("csrf") == "t2"
        return (200, "ok" if state["in"] else OTP_PAGE)

    return {
        "https://club.test/login": LOGIN_PAGE,
        "https://club.test/": lambda: HOME_OPEN if state.get("in") else LOGIN_PAGE,
        "post:https://club.test/login": login,
        "post:https://club.test/2fa": second,
    }


def test_a_form_login_with_a_second_factor_submits_the_current_code(monkeypatch):
    monkeypatch.setattr(runtime, "totp", lambda secret: "123456")
    state = {}
    collector = Collector(FakeFetcher(club(state)), form_schema(), sleep=no_sleep, credentials={"username": "bob", "password": "pw", "totp": RFC_SECRET})
    assert run_async(collector.login()) is True
    assert state["in"] and state["codes"] == ["123456"]


def test_a_second_factor_without_a_secret_says_what_is_missing():
    collector = Collector(FakeFetcher(club({})), form_schema(), sleep=no_sleep, credentials={"username": "bob", "password": "pw"})
    with pytest.raises(SecondFactorRequired, match="authenticator secret"):
        run_async(collector.login())


def test_a_rejected_code_is_reported_not_retried(monkeypatch):
    monkeypatch.setattr(runtime, "totp", lambda secret: "000000")
    state = {}
    collector = Collector(FakeFetcher(club(state)), form_schema(), sleep=no_sleep, credentials={"username": "bob", "password": "pw", "totp": RFC_SECRET})
    with pytest.raises(SecondFactorRequired, match="rejected the one-time code"):
        run_async(collector.login())
    assert state["codes"] == ["000000"]


def test_a_login_without_a_second_factor_is_unchanged():
    state = {}
    collector = Collector(FakeFetcher(club(state, challenge=False)), form_schema(), sleep=no_sleep, credentials={"username": "bob", "password": "pw"})
    assert run_async(collector.login()) is True and state["in"]


def test_wrong_password_still_fails_as_before():
    collector = Collector(FakeFetcher(club({})), form_schema(), sleep=no_sleep, credentials={"username": "bob", "password": "nope"})
    with pytest.raises(LoginFailed, match="rejected the stored credentials"):
        run_async(collector.login())


def test_a_captcha_on_the_login_page_is_never_attempted():
    routes = club({})
    routes["https://club.test/login"] = LOGIN_PAGE.replace("<form", '<div class="g-recaptcha" data-sitekey="k"></div><form', 1)
    fetcher = FakeFetcher(routes)
    collector = Collector(fetcher, form_schema(), sleep=no_sleep, credentials={"username": "bob", "password": "pw"})
    with pytest.raises(CaptchaRequired, match="captcha"):
        run_async(collector.login())
    assert fetcher.posts == []


def test_a_learned_captcha_flag_stops_before_any_request():
    schema = form_schema()
    schema["login"]["captcha"] = True
    fetcher = FakeFetcher({})
    with pytest.raises(CaptchaRequired):
        run_async(Collector(fetcher, schema, sleep=no_sleep, credentials={"username": "bob", "password": "pw"}).login())
    assert fetcher.requests == [] and fetcher.posts == []


def json_schema(**login):
    return {
        "community": "api",
        "origin": "api.test",
        "base_url": "https://api.test/",
        "feeds": [],
        "limits": {"delay_seconds": 0, "retries": 1},
        "login": {"mode": "json", "url": "https://api.test/api/login", "username_field": "email", "password_field": "password", "identity": "email", **login},
    }


def api(state, challenge=False):
    def login(body):
        state["bodies"] = state.get("bodies", []) + [dict(body)]
        if body.get("email") != "bob@x.test" or body.get("password") != "pw":
            return (401, {"error": "bad credentials"})
        if challenge and body.get("otp") != "123456":
            return (200, {"otp_required": True}) if "otp" not in body else (401, {"error": "invalid verification code"})
        return (200, {"data": {"token": "tok-1"}})

    return {"post:https://api.test/api/login": login, "https://api.test/api/me": {"ok": True}}


def test_a_json_login_stores_the_token_and_sends_it_on_later_requests():
    state = {}
    fetcher = FakeFetcher(api(state))
    collector = Collector(fetcher, json_schema(token_path="$.data.token"), sleep=no_sleep, credentials={"email": "bob@x.test", "password": "pw"})
    assert run_async(collector.login()) is True
    assert state["bodies"] == [{"email": "bob@x.test", "password": "pw"}]
    run_async(collector.fetch("https://api.test/api/me"))
    assert fetcher.get_headers[-1] == {"Authorization": "Bearer tok-1"}


def test_a_json_login_without_a_token_relies_on_the_session_cookie():
    collector = Collector(FakeFetcher(api({})), json_schema(), sleep=no_sleep, credentials={"email": "bob@x.test", "password": "pw"})
    assert run_async(collector.login()) is True and collector.headers == {}


def test_a_json_login_with_a_missing_token_or_bad_password_fails():
    with pytest.raises(LoginFailed, match="without the expected token"):
        run_async(Collector(FakeFetcher(api({})), json_schema(token_path="$.nope"), sleep=no_sleep, credentials={"email": "bob@x.test", "password": "pw"}).login())
    with pytest.raises(LoginFailed, match="rejected the stored credentials"):
        run_async(Collector(FakeFetcher(api({})), json_schema(), sleep=no_sleep, credentials={"email": "bob@x.test", "password": "bad"}).login())


def test_a_json_login_answers_the_code_challenge(monkeypatch):
    monkeypatch.setattr(runtime, "totp", lambda secret: "123456")
    state = {}
    collector = Collector(FakeFetcher(api(state, challenge=True)), json_schema(token_path="$.data.token"), sleep=no_sleep, credentials={"email": "bob@x.test", "password": "pw", "totp": RFC_SECRET})
    assert run_async(collector.login()) is True
    assert [("otp" in body) for body in state["bodies"]] == [False, True]
    with pytest.raises(SecondFactorRequired, match="authenticator secret"):
        run_async(Collector(FakeFetcher(api({}, challenge=True)), json_schema(), sleep=no_sleep, credentials={"email": "bob@x.test", "password": "pw"}).login())


def test_code_challenge_detection():
    assert code_challenge({"otp_required": True}, "", 200)
    assert code_challenge({"Two_Factor": 1}, "", 200)
    assert code_challenge(None, "Please enter your verification code", 401)
    assert not code_challenge({"token": "x"}, "", 200)
    assert not code_challenge(None, "Please enter your verification code", 200)


def test_the_learner_finds_a_json_login_and_describes_it():
    state = {}
    routes = api(state)
    routes["post:https://api.test/api/auth/login"] = (404, "nope")
    learner = LoginLearner(FakeFetcher(routes))
    spec = run_async(learner.find_json("https://api.test/", {"email": "bob@x.test", "password": "pw"}))
    assert spec == {
        "mode": "json", "url": "https://api.test/api/login", "username_field": "email", "password_field": "password",
        "identity": "email", "token_path": "$.data.token", "token_scheme": "Bearer",
    }
    token_routes = {"post:https://api.test/api/login": lambda body: (200, {"access_token": "t"}) if body.get("username") == "bob" else (401, {"error": "x"})}
    spec = run_async(LoginLearner(FakeFetcher(token_routes)).find_json("https://api.test/", {"username": "bob", "password": "pw"}))
    assert spec["token_path"] == "$.access_token" and spec["username_field"] == "username"


def test_the_learner_gives_up_quietly_and_within_its_attempt_cap():
    fetcher = FakeFetcher({"post:https://api.test/api/login": (401, {"error": "no"})})
    assert run_async(LoginLearner(fetcher).find_json("https://api.test/", {"username": "bob", "password": "x"})) is None
    assert len(fetcher.posts) <= 12
    html_only = FakeFetcher({"post:https://api.test/api/login": (200, "<html>not json</html>")})
    assert run_async(LoginLearner(html_only).find_json("https://api.test/", {"username": "bob", "password": "x"})) is None