split-trainer/server/tests/test_api.py

"""Интеграционные тесты API: реальный HTTP-сервер, SQLite во временной папке, почта в памяти."""
import http.client
import json
import os
import re
import sqlite3
import tempfile
import threading
import unittest
from unittest import mock

from splitapi import security
from splitapi.config import from_env
from splitapi.http_api import is_trusted_proxy_peer, make_server, normalize_ip
from splitapi.mailer import MailQueueFull, MemoryMailer
from splitapi.ratelimit import RateLimiter
from splitapi.service import Service
from splitapi.store import Store

PASSWORD = "correct horse battery"
OTHER_PASSWORD = "another staple pony"
_ORIGINAL_N = security.SCRYPT_N
WRONG = "definitely wrong pass"


def setUpModule():
    security.SCRYPT_N = 2**10        # быстрые хеши в тестах; боевое значение проверено в test_security


def tearDownModule():
    security.SCRYPT_N = _ORIGINAL_N


class Clock:
    def __init__(self):
        self.now = 1_800_000_000.0

    def __call__(self):
        return self.now

    def advance(self, seconds):
        self.now += seconds


class Resp:
    def __init__(self, status, headers, body):
        self.status, self.headers, self.text = status, headers, body
        self.set_cookies = []
        try:
            self.json = json.loads(body) if body else None
        except ValueError:
            self.json = None

    def header(self, name):
        return self.headers.get(name)


class ApiTestCase(unittest.TestCase):
    public_url = "http://127.0.0.1:8080"
    env_extra: dict = {}

    def make_mailer(self):
        return MemoryMailer()

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.db_path = os.path.join(self.tmp.name, "api.sqlite3")
        self.cfg = from_env({"PUBLIC_URL": self.public_url, "DB_PATH": self.db_path, "MAIL_BACKEND": "console", **self.env_extra})
        self.clock = Clock()
        self.mail = self.make_mailer()
        self.service = Service(self.cfg, Store(self.db_path), self.mail, RateLimiter(clock=self.clock), clock=self.clock)
        self.server = make_server(self.cfg, self.service, host="127.0.0.1", port=0)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(target=lambda: self.server.serve_forever(poll_interval=0.01), daemon=True)
        self.thread.start()
        self.addCleanup(self._stop)
        self.cookie = None

    def _stop(self):
        self.server.shutdown()
        self.server.server_close()

    # --- помощники ---
    def call(self, method, path, body=None, headers=None, raw=None, csrf=True, cookie="auto"):
        hdrs = {}
        if method not in ("GET", "HEAD", "OPTIONS") and csrf:
            hdrs.update({"Content-Type": "application/json", "X-Requested-With": "split-trainer", "Origin": self.cfg.origin})
        if cookie == "auto":
            cookie = self.cookie
        if cookie:
            hdrs["Cookie"] = f"{self.cfg.cookie_name}={cookie}"
        hdrs.update(headers or {})
        payload = raw if raw is not None else (json.dumps(body).encode() if body is not None else None)
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        try:
            conn.request(method, path, body=payload, headers=hdrs)
            r = conn.getresponse()
            resp = Resp(r.status, {k: v for k, v in r.getheaders()}, r.read().decode("utf-8"))
            resp.set_cookies = [v for k, v in r.getheaders() if k.lower() == "set-cookie"]
            return resp
        finally:
            conn.close()

    def register(self, email="user@example.com", **extra):
        return self.call("POST", "/api/register", {"email": email, **extra})

    def mails_to(self, email):
        return [m for m in self.mail.sent if m.to == email]

    def token_from(self, email, kind, index=-1):
        mail = self.mails_to(email)[index]
        match = re.search(rf"#{kind}=([A-Za-z0-9_-]+)", mail.body)
        self.assertIsNotNone(match, f"в письме нет ссылки {kind}: {mail.body}")
        return match.group(1)

    def confirm(self, email="user@example.com", password=PASSWORD):
        return self.call("POST", "/api/verify", {"token": self.token_from(email, "verify"), "password": password})

    def login(self, email="user@example.com", password=PASSWORD, remember=True, headers=None):
        r = self.call("POST", "/api/login", {"email": email, "password": password}, cookie=None, headers=headers)
        if r.status == 200 and remember:
            self.cookie = re.match(rf"{re.escape(self.cfg.cookie_name)}=([^;]+)", r.set_cookies[0]).group(1)
        return r

    def whoami(self, cookie="auto"):
        r = self.call("GET", "/api/me", cookie=cookie)
        self.assertEqual(r.status, 200)           # гостю тоже 200: так в консоли браузера нет «ошибок» при каждом заходе
        return r.json["email"]

    def make_user(self, email="user@example.com", password=PASSWORD):
        self.assertEqual(self.register(email).status, 200)
        self.assertEqual(self.confirm(email, password).status, 200)
        self.assertEqual(self.login(email, password).status, 200)

    def fail_logins(self, count, email="user@example.com", headers=None):
        for _ in range(count):
            self.assertEqual(self.login(email, WRONG, remember=False, headers=headers).status, 401)


class BasicTests(ApiTestCase):
    def test_health(self):
        r = self.call("GET", "/api/health")
        self.assertEqual((r.status, r.json), (200, {"ok": True}))

    def test_unknown_path_and_wrong_method(self):
        self.assertEqual(self.call("GET", "/api/nope").status, 404)
        r = self.call("GET", "/api/login")
        self.assertEqual(r.status, 405)
        self.assertIn("POST", r.header("Allow"))
        self.assertEqual(self.call("POST", "/api/health", {}).status, 405)

    def test_head_and_options_are_not_supported(self):
        self.assertEqual(self.call("HEAD", "/api/health").status, 405)
        self.assertEqual(self.call("OPTIONS", "/api/login").status, 405)

    def test_responses_are_not_cacheable_and_nosniff(self):
        r = self.call("GET", "/api/health")
        self.assertEqual(r.header("Cache-Control"), "no-store")
        self.assertEqual(r.header("X-Content-Type-Options"), "nosniff")
        self.assertTrue(r.header("Content-Type").startswith("application/json"))


class RegistrationTests(ApiTestCase):
    def test_register_sends_confirmation_link_and_creates_no_session(self):
        r = self.register()
        self.assertEqual((r.status, r.json["ok"]), (200, True))
        self.assertEqual(len(self.mails_to("user@example.com")), 1)
        self.assertIn("http://127.0.0.1:8080/#verify=", self.mails_to("user@example.com")[0].body)
        self.assertEqual(r.set_cookies, [])

    def test_registration_never_sets_a_password(self):
        """Пароль выбирает тот, у кого есть доступ к ящику, поэтому подменить его чужой регистрацией нельзя."""
        self.register(password="attackers-password-1")
        self.assertEqual(self.login(password="attackers-password-1", remember=False).status, 401)
        self.assertEqual(self.confirm(password=OTHER_PASSWORD).status, 200)
        self.assertEqual(self.login(password="attackers-password-1", remember=False).status, 401)
        self.assertEqual(self.login(password=OTHER_PASSWORD).status, 200)

    def test_cannot_login_before_confirmation(self):
        self.register()
        r = self.login()
        self.assertEqual((r.status, r.json["error"]), (401, "invalid_credentials"))

    def test_confirmation_creates_account_with_chosen_password_without_logging_in(self):
        self.register()
        r = self.confirm()
        self.assertEqual(r.status, 200)
        self.assertEqual(r.set_cookies, [])
        self.assertEqual(self.login().status, 200)

    def test_confirmation_token_is_single_use(self):
        self.register()
        token = self.token_from("user@example.com", "verify")
        self.assertEqual(self.call("POST", "/api/verify", {"token": token, "password": PASSWORD}).status, 200)
        r = self.call("POST", "/api/verify", {"token": token, "password": OTHER_PASSWORD})
        self.assertEqual((r.status, r.json["error"]), (400, "invalid_token"))
        self.assertEqual(self.login(password=OTHER_PASSWORD, remember=False).status, 401)

    def test_confirmation_token_expires(self):
        self.register()
        token = self.token_from("user@example.com", "verify")
        self.clock.advance(24 * 3600 + 1)
        self.assertEqual(self.call("POST", "/api/verify", {"token": token, "password": PASSWORD}).json["error"], "invalid_token")

    def test_garbage_tokens_are_rejected(self):
        for token in ["", "short", None, 123, "x" * 500, "../../etc/passwd", {"a": 1}]:
            with self.subTest(token=token):
                self.assertEqual(self.call("POST", "/api/verify", {"token": token, "password": PASSWORD}).status, 400)

    def test_weak_password_at_confirmation_does_not_burn_the_link(self):
        self.register()
        token = self.token_from("user@example.com", "verify")
        for bad, code in [("short", "password_too_short"), ("password", "password_weak"), ("x" * 200, "password_too_long"),
                          ("user@example.com", "password_weak"), (None, "password_too_short")]:
            with self.subTest(bad=bad):
                self.assertEqual(self.call("POST", "/api/verify", {"token": token, "password": bad}).json["error"], code)
        self.assertEqual(self.call("POST", "/api/verify", {"token": token, "password": PASSWORD}).status, 200)

    def test_lone_surrogate_password_is_a_clean_400_not_a_500(self):
        self.register()
        token = self.token_from("user@example.com", "verify")
        r = self.call("POST", "/api/verify", raw=b'{"token": "%s", "password": "\\ud800\\ud800\\ud800\\ud800\\ud800\\ud800\\ud800\\ud800"}' % token.encode())
        self.assertEqual((r.status, r.json["error"]), (400, "password_invalid"))
        self.assertEqual(self.call("POST", "/api/verify", {"token": token, "password": PASSWORD}).status, 200)

    def test_latest_registration_link_wins_and_older_dies(self):
        self.register()
        old = self.token_from("user@example.com", "verify")
        self.clock.advance(1)
        self.register()
        self.assertEqual(self.call("POST", "/api/verify", {"token": old, "password": PASSWORD}).status, 400)
        self.assertEqual(self.confirm().status, 200)

    def test_registering_existing_email_is_indistinguishable_and_sends_notice_only(self):
        fresh = self.register("fresh@example.com")
        self.make_user("user@example.com")
        count_before = len(self.mails_to("user@example.com"))
        again = self.register("user@example.com")
        self.assertEqual((again.status, again.json), (fresh.status, fresh.json))
        notice = self.mails_to("user@example.com")[count_before:]
        self.assertEqual(len(notice), 1)
        self.assertNotIn("#verify=", notice[0].body)
        self.assertEqual(self.login(remember=False).status, 200)

    def test_account_taken_while_link_was_pending_cannot_be_overwritten(self):
        self.register()
        stale = self.token_from("user@example.com", "verify")
        self.register()
        self.assertEqual(self.confirm(password=PASSWORD).status, 200)
        self.assertEqual(self.call("POST", "/api/verify", {"token": stale, "password": OTHER_PASSWORD}).status, 400)
        self.assertEqual(self.login(password=PASSWORD, remember=False).status, 200)

    def test_invalid_emails_are_rejected_without_sending_mail(self):
        for bad in ["not-an-email", "a\n@example.com", "user@example\n.com", "a%b@example.com", ["a@example.com"], None, 5]:
            with self.subTest(bad=bad):
                self.assertEqual(self.register(bad).json["error"], "invalid_email")
                self.clock.advance(1000)                        # не упираемся в лимит регистраций с одного IP
        self.assertEqual(self.mail.sent, [])

    def test_email_is_normalized(self):
        self.register("  User@Example.COM ")
        self.assertEqual(len(self.mails_to("user@example.com")), 1)
        self.confirm("user@example.com")
        self.assertEqual(self.login("USER@example.com").status, 200)

    def test_registration_rate_limit_per_ip(self):
        statuses = [self.register(f"u{i}@example.com").status for i in range(7)]
        self.assertEqual(statuses[:5], [200] * 5)
        self.assertEqual(statuses[5:], [429, 429])
        self.assertIsNotNone(self.register("late@example.com").header("Retry-After"))

    def test_per_email_limit_stays_silent(self):
        for _ in range(3):
            self.assertEqual(self.register("same@example.com").status, 200)
            self.clock.advance(1000)                   # чтобы не упереться в лимит по IP
        before = len(self.mails_to("same@example.com"))
        r = self.register("same@example.com")
        self.assertEqual((r.status, r.json["ok"]), (200, True))
        self.assertEqual(len(self.mails_to("same@example.com")), before)

    def test_expired_registrations_are_purged_periodically_not_on_every_request(self):
        self.register("old@example.com")
        self.clock.advance(24 * 3600 + 1)
        self.register("new1@example.com")                      # первая чистка
        self.assertEqual(Store(self.db_path).count("registrations"), 1)
        self.clock.advance(1)
        self.register("new2@example.com")
        self.assertEqual(Store(self.db_path).count("registrations"), 2)


class RegistrationClosedTests(ApiTestCase):
    env_extra = {"ALLOW_REGISTRATION": "false"}

    def test_registration_can_be_disabled(self):
        r = self.register()
        self.assertEqual((r.status, r.json["error"]), (403, "registration_closed"))
        self.assertEqual(self.mail.sent, [])


class MailQueueFullMailer:
    def send(self, mail):
        raise MailQueueFull


class MailOverloadTests(ApiTestCase):
    def make_mailer(self):
        return MailQueueFullMailer()

    def test_overloaded_mail_queue_is_reported_instead_of_pretending_the_mail_was_sent(self):
        r = self.register()
        self.assertEqual((r.status, r.json["error"]), (503, "busy"))
        self.assertIsNotNone(r.header("Retry-After"))

    def test_forgot_does_not_reveal_which_addresses_exist_when_the_queue_is_full(self):
        """При переполненной очереди известный адрес не должен отвечать иначе, чем неизвестный."""
        Store(self.db_path).create_user("known@example.com", security.hash_password(PASSWORD), self.clock())
        known = self.call("POST", "/api/forgot", {"email": "known@example.com"})
        unknown = self.call("POST", "/api/forgot", {"email": "ghost@example.com"})
        self.assertEqual((known.status, known.json), (unknown.status, unknown.json))
        self.assertEqual(known.status, 200)

    def test_password_change_notice_failing_does_not_undo_the_reset(self):
        Store(self.db_path).create_user("known@example.com", security.hash_password(PASSWORD), self.clock())
        user = Store(self.db_path).user_by_email("known@example.com")
        Store(self.db_path).save_reset_token(user["id"], security.hash_token("T" * 43), self.clock() + 3600)
        r = self.call("POST", "/api/reset", {"token": "T" * 43, "password": OTHER_PASSWORD})
        self.assertEqual(r.status, 200)
        self.assertEqual(self.login("known@example.com", OTHER_PASSWORD, remember=False).status, 200)


class LoginTests(ApiTestCase):
    def test_login_sets_hardened_cookie_and_me_works(self):
        self.make_user()
        r = self.call("POST", "/api/login", {"email": "user@example.com", "password": PASSWORD}, cookie=None)
        cookie = r.set_cookies[0]
        for flag in ("HttpOnly", "SameSite=Lax", "Path=/", "Max-Age=2592000"):
            self.assertIn(flag, cookie)
        self.assertNotIn("Secure", cookie)                      # PUBLIC_URL по http
        self.assertEqual(self.whoami(), "user@example.com")

    def test_wrong_password_and_unknown_user_look_identical(self):
        self.make_user()
        wrong = self.login(password=WRONG, remember=False)
        unknown = self.login(email="ghost@example.com", remember=False)
        self.assertEqual((wrong.status, wrong.json), (unknown.status, unknown.json))
        self.assertEqual(wrong.status, 401)

    def test_me_reports_guest_with_null_email(self):
        self.assertIsNone(self.whoami(cookie=None))
        self.assertIsNone(self.whoami(cookie="x" * 43))

    def test_garbage_and_duplicate_cookies_mean_guest_not_an_error(self):
        self.make_user()
        for header in ["session=;;;====", "\x01\x02", "=", ";", f"{self.cfg.cookie_name}={self.cookie}; {self.cfg.cookie_name}=other"]:
            with self.subTest(header=header):
                r = self.call("GET", "/api/me", cookie=None, headers={"Cookie": header})
                self.assertEqual((r.status, r.json["email"]), (200, None))

    def test_logout_invalidates_session_and_clears_cookie(self):
        self.make_user()
        r = self.call("POST", "/api/logout", {})
        self.assertEqual(r.status, 200)
        self.assertIn("Max-Age=0", r.set_cookies[0])
        self.assertIsNone(self.whoami())

    def test_session_expires(self):
        self.make_user()
        self.clock.advance(30 * 24 * 3600 + 1)
        self.assertIsNone(self.whoami())

    def test_failed_logins_per_ip_are_limited_and_recover(self):
        for i in range(20):
            self.assertEqual(self.login(f"x{i}@example.com", WRONG, remember=False).status, 401)
        r = self.login("y@example.com", WRONG, remember=False)
        self.assertEqual(r.status, 429)
        self.assertGreaterEqual(int(r.header("Retry-After")), 1)
        self.clock.advance(601)
        self.assertEqual(self.login("y@example.com", WRONG, remember=False).status, 401)

    def test_successful_logins_do_not_use_up_the_ip_allowance(self):
        """Класс или семья за одним NAT входят много раз подряд: успехи не считаются."""
        self.make_user()
        for _ in range(30):
            self.assertEqual(self.login(remember=False).status, 200)

    def test_password_guessing_for_one_account_from_one_ip_is_blocked_even_with_the_right_password(self):
        self.make_user()
        self.fail_logins(5)
        self.assertEqual(self.login(remember=False).status, 429)

    def test_successful_logins_do_not_count_toward_the_account_lockout(self):
        self.make_user()
        self.fail_logins(4)
        for _ in range(3):
            self.assertEqual(self.login(remember=False).status, 200)
        self.fail_logins(1)
        self.assertEqual(self.login(remember=False).status, 429)

    def test_lockout_expires(self):
        self.make_user()
        self.fail_logins(5)
        self.clock.advance(601)
        self.assertEqual(self.login(remember=False).status, 200)

    def test_busy_hasher_gives_503_with_retry_after(self):
        self.make_user()
        with mock.patch.object(security, "_slots", threading.BoundedSemaphore(0)), mock.patch.object(security, "SLOT_WAIT", 0.05):
            r = self.login(remember=False)
        self.assertEqual((r.status, r.json["error"]), (503, "busy"))
        self.assertIsNotNone(r.header("Retry-After"))

    def test_lone_surrogate_password_at_login_is_a_clean_401(self):
        self.make_user()
        r = self.call("POST", "/api/login", raw=b'{"email": "user@example.com", "password": "\\ud800\\ud800\\ud800"}', cookie=None)
        self.assertEqual((r.status, r.json["error"]), (401, "invalid_credentials"))

    def test_rehash_never_overwrites_a_password_reset_that_happened_meanwhile(self):
        """Старый (скомпрометированный) пароль не должен вернуться из-за перехеширования, начавшегося до сброса."""
        self.make_user()
        store = Store(self.db_path)
        uid = store.user_by_email("user@example.com")["id"]
        real_hash = security.hash_password

        def racing_hash(password, n=None):
            store.set_password(uid, "HASH-AFTER-RESET")        # владелец успел сменить пароль, пока считался новый хеш
            return real_hash(password, n)

        with mock.patch.object(security, "needs_rehash", return_value=True), mock.patch.object(security, "hash_password", racing_hash):
            r = self.login(remember=False)
        self.assertEqual((r.status, r.json["error"]), (401, "invalid_credentials"))
        self.assertEqual(store.user_by_email("user@example.com")["pw_hash"], "HASH-AFTER-RESET")
        self.assertIsNone(self.whoami(cookie=None))

    def test_password_hash_is_upgraded_on_login(self):
        self.make_user()
        old_n = security.SCRYPT_N
        security.SCRYPT_N = 2**11
        try:
            self.assertEqual(self.login(remember=False).status, 200)
            stored = Store(self.db_path).user_by_email("user@example.com")["pw_hash"]
            self.assertTrue(stored.startswith("scrypt$2048$"))
            self.assertEqual(self.login(remember=False).status, 200)                      # новым хешем тоже входит
        finally:
            security.SCRYPT_N = old_n


class PasswordResetTests(ApiTestCase):
    def test_forgot_for_unknown_email_is_indistinguishable_and_sends_nothing(self):
        self.make_user()
        known = self.call("POST", "/api/forgot", {"email": "user@example.com"})
        unknown = self.call("POST", "/api/forgot", {"email": "ghost@example.com"})
        self.assertEqual((known.status, known.json), (unknown.status, unknown.json))
        self.assertEqual(self.mails_to("ghost@example.com"), [])
        self.assertIn("#reset=", self.mails_to("user@example.com")[-1].body)

    def test_forgot_rejects_malformed_email(self):
        self.assertEqual(self.call("POST", "/api/forgot", {"email": "nope"}).json["error"], "invalid_email")

    def test_full_reset_flow(self):
        self.make_user()
        old_cookie = self.cookie
        self.call("POST", "/api/forgot", {"email": "user@example.com"})
        token = self.token_from("user@example.com", "reset")
        r = self.call("POST", "/api/reset", {"token": token, "password": OTHER_PASSWORD}, cookie=None)
        self.assertEqual(r.status, 200)
        self.assertEqual(self.login(password=PASSWORD, remember=False).status, 401)
        self.assertEqual(self.login(password=OTHER_PASSWORD).status, 200)
        self.assertIsNone(self.whoami(cookie=old_cookie))                                  # старые сессии отозваны
        self.assertIn("пароль", self.mails_to("user@example.com")[-1].subject.lower())

    def test_reset_token_is_single_use(self):
        self.make_user()
        self.call("POST", "/api/forgot", {"email": "user@example.com"})
        token = self.token_from("user@example.com", "reset")
        self.assertEqual(self.call("POST", "/api/reset", {"token": token, "password": OTHER_PASSWORD}).status, 200)
        r = self.call("POST", "/api/reset", {"token": token, "password": "third password!"})
        self.assertEqual((r.status, r.json["error"]), (400, "invalid_token"))

    def test_bad_new_password_does_not_burn_the_token(self):
        self.make_user()
        self.call("POST", "/api/forgot", {"email": "user@example.com"})
        token = self.token_from("user@example.com", "reset")
        self.assertEqual(self.call("POST", "/api/reset", {"token": token, "password": "123"}).json["error"], "password_too_short")
        surrogate = self.call("POST", "/api/reset", raw=b'{"token": "%s", "password": "\\ud800\\ud800\\ud800\\ud800\\ud800\\ud800\\ud800\\ud800"}' % token.encode())
        self.assertEqual((surrogate.status, surrogate.json["error"]), (400, "password_invalid"))
        self.assertEqual(self.call("POST", "/api/reset", {"token": token, "password": OTHER_PASSWORD}).status, 200)

    def test_reset_token_expires_after_one_hour(self):
        self.make_user()
        self.call("POST", "/api/forgot", {"email": "user@example.com"})
        token = self.token_from("user@example.com", "reset")
        self.clock.advance(3601)
        self.assertEqual(self.call("POST", "/api/reset", {"token": token, "password": OTHER_PASSWORD}).json["error"], "invalid_token")

    def test_new_request_invalidates_previous_link(self):
        self.make_user()
        self.call("POST", "/api/forgot", {"email": "user@example.com"})
        first = self.token_from("user@example.com", "reset")
        self.clock.advance(5)
        self.call("POST", "/api/forgot", {"email": "user@example.com"})
        self.assertEqual(self.call("POST", "/api/reset", {"token": first, "password": OTHER_PASSWORD}).status, 400)
        self.assertEqual(self.call("POST", "/api/reset", {"token": self.token_from("user@example.com", "reset"), "password": OTHER_PASSWORD}).status, 200)

    def test_forgot_rate_limit_per_email_is_silent(self):
        self.make_user()
        for _ in range(5):
            self.assertEqual(self.call("POST", "/api/forgot", {"email": "user@example.com"}).status, 200)
        resets = [m for m in self.mails_to("user@example.com") if "#reset=" in m.body]
        self.assertEqual(len(resets), 3)

    def test_forgot_rate_limit_per_ip(self):
        statuses = [self.call("POST", "/api/forgot", {"email": f"a{i}@example.com"}).status for i in range(7)]
        self.assertEqual(statuses[-1], 429)

    def test_reset_cannot_be_used_with_a_verify_token(self):
        self.register()
        token = self.token_from("user@example.com", "verify")
        self.assertEqual(self.call("POST", "/api/reset", {"token": token, "password": OTHER_PASSWORD}).status, 400)

    def test_session_created_by_a_concurrent_login_does_not_survive_the_reset(self):
        """Вход, начавшийся со старым паролем, не должен пережить сброс пароля."""
        self.make_user()
        stale_hash = Store(self.db_path).user_by_email("user@example.com")["pw_hash"]
        self.call("POST", "/api/forgot", {"email": "user@example.com"})
        self.call("POST", "/api/reset", {"token": self.token_from("user@example.com", "reset"), "password": OTHER_PASSWORD})
        user = Store(self.db_path).user_by_email("user@example.com")
        self.assertFalse(Store(self.db_path).create_session(user["id"], "late-session-hash", 2**40, 1, expect_pw_hash=stale_hash))


class DeleteAccountTests(ApiTestCase):
    def test_requires_login(self):
        self.assertEqual(self.call("POST", "/api/delete-account", {"password": PASSWORD}, cookie=None).status, 401)

    def test_wrong_password_keeps_the_account(self):
        self.make_user()
        r = self.call("POST", "/api/delete-account", {"password": WRONG})
        self.assertEqual((r.status, r.json["error"]), (401, "wrong_password"))
        self.assertEqual(self.login(remember=False).status, 200)

    def test_deletes_account_progress_and_sessions(self):
        self.make_user()
        self.call("PUT", "/api/state", {"version": 0, "data": {"progress": {"ru-A1": {"best": 1, "acc": 1, "stars": 1, "tries": 1}}}})
        r = self.call("POST", "/api/delete-account", {"password": PASSWORD})
        self.assertEqual(r.status, 200)
        self.assertIn("Max-Age=0", r.set_cookies[0])
        self.assertIsNone(self.whoami())
        self.assertEqual(self.login(remember=False).status, 401)
        store = Store(self.db_path)
        self.assertEqual((store.count("users"), store.count("user_state"), store.count("sessions")), (0, 0, 0))

    def test_password_guessing_here_is_limited_too(self):
        self.make_user()
        for _ in range(5):
            self.assertEqual(self.call("POST", "/api/delete-account", {"password": WRONG}).status, 401)
        self.assertEqual(self.call("POST", "/api/delete-account", {"password": PASSWORD}).status, 429)

    def test_email_can_register_again_after_deletion(self):
        self.make_user()
        self.call("POST", "/api/delete-account", {"password": PASSWORD})
        self.cookie = None
        self.make_user(password=OTHER_PASSWORD)
        self.assertEqual(self.whoami(), "user@example.com")


class RequestHardeningTests(ApiTestCase):
    def test_missing_csrf_header_is_rejected(self):
        r = self.call("POST", "/api/login", {"email": "a@example.com", "password": "x" * 10},
                      headers={"Content-Type": "application/json", "Origin": self.cfg.origin}, csrf=False)
        self.assertEqual((r.status, r.json["error"]), (403, "csrf"))

    def test_foreign_origin_is_rejected_and_logged_with_expected_value(self):
        with self.assertLogs("splitapi", level="WARNING") as logs:
            r = self.call("POST", "/api/login", {"email": "a@example.com", "password": "x" * 10}, headers={"Origin": "https://evil.example"})
        self.assertEqual((r.status, r.json["error"]), (403, "csrf"))
        text = "\n".join(logs.output)
        self.assertIn("https://evil.example", text)
        self.assertIn(self.cfg.origin, text)

    def test_origin_null_is_rejected(self):
        r = self.call("POST", "/api/login", {"email": "a@example.com", "password": "x" * 10}, headers={"Origin": "null"})
        self.assertEqual(r.status, 403)

    def test_request_without_origin_but_with_custom_header_is_allowed(self):
        hdrs = {"Content-Type": "application/json", "X-Requested-With": "split-trainer"}
        r = self.call("POST", "/api/login", {"email": "a@example.com", "password": "x" * 10}, headers=hdrs, csrf=False)
        self.assertEqual(r.status, 401)

    def test_wrong_content_type_is_rejected(self):
        r = self.call("POST", "/api/login", raw=b'{"email":"a@example.com"}', headers={"Content-Type": "text/plain"})
        self.assertEqual(r.status, 415)

    def test_invalid_json_and_non_object_bodies(self):
        self.assertEqual(self.call("POST", "/api/login", raw=b"{not json").json["error"], "invalid_json")
        self.assertEqual(self.call("POST", "/api/login", raw=b"[1,2,3]").json["error"], "invalid_json")
        self.assertEqual(self.call("POST", "/api/login", raw=b"").status, 400)
        self.assertEqual(self.call("POST", "/api/login", raw=b'{"a": NaN}').status, 400)

    def test_oversized_body_is_rejected_before_reading(self):
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        conn.putrequest("POST", "/api/login")
        for k, v in {"Content-Type": "application/json", "X-Requested-With": "split-trainer", "Content-Length": "5000000"}.items():
            conn.putheader(k, v)
        conn.endheaders()
        self.assertEqual(conn.getresponse().status, 413)
        conn.close()

    def test_errors_do_not_leak_internals(self):
        r = self.call("POST", "/api/login", raw=b"\xff\xfe")
        self.assertNotIn("Traceback", r.text)
        self.assertLess(r.status, 500)


class StateTests(ApiTestCase):
    GOOD = {"progress": {"ru-A1": {"best": 100, "acc": 95, "stars": 1, "tries": 1}}, "history": [], "keys": {"ru": {}, "en": {}}}

    def test_state_requires_login(self):
        self.assertEqual(self.call("GET", "/api/state", cookie=None).status, 401)
        self.assertEqual(self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD}, cookie=None).status, 401)

    def test_roundtrip_and_versioning(self):
        self.make_user()
        first = self.call("GET", "/api/state")
        self.assertEqual((first.json["version"], first.json["data"]["progress"]), (0, {}))
        saved = self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD})
        self.assertEqual((saved.status, saved.json["version"]), (200, 1))
        got = self.call("GET", "/api/state").json
        self.assertEqual((got["version"], got["data"]["progress"]["ru-A1"]["best"]), (1, 100))

    def test_state_response_names_the_account_so_a_stale_tab_can_notice_a_switch(self):
        self.make_user("a@example.com")
        self.assertEqual(self.call("GET", "/api/state").json["email"], "a@example.com")

    def test_saving_on_behalf_of_another_account_is_refused(self):
        """Вкладка, открытая под аккаунтом A, не должна записать данные в аккаунт B после смены входа в другой вкладке."""
        self.make_user("b@example.com")
        r = self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD, "email": "a@example.com"})
        self.assertEqual((r.status, r.json["error"]), (409, "account_mismatch"))
        self.assertEqual(self.call("GET", "/api/state").json["data"]["progress"], {})
        ok = self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD, "email": "B@Example.com"})
        self.assertEqual(ok.status, 200)

    def test_saving_without_an_email_still_works_for_older_clients(self):
        self.make_user()
        self.assertEqual(self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD}).status, 200)

    def test_progress_can_be_cleared_by_saving_an_empty_state(self):
        self.make_user()
        self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD})
        self.assertEqual(self.call("PUT", "/api/state", {"version": 1, "data": {}}).status, 200)
        self.assertEqual(self.call("GET", "/api/state").json["data"]["progress"], {})

    def test_stale_version_gets_conflict_with_current_state(self):
        self.make_user()
        self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD})
        r = self.call("PUT", "/api/state", {"version": 0, "data": {"progress": {}, "history": [], "keys": {}}})
        self.assertEqual((r.status, r.json["error"], r.json["version"]), (409, "version_conflict", 1))
        self.assertIn("ru-A1", r.json["data"]["progress"])

    def test_parallel_writes_with_the_same_version_let_exactly_one_through(self):
        self.make_user()
        statuses, gate = [], threading.Barrier(2)

        def writer():
            gate.wait()
            statuses.append(self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD}).status)

        threads = [threading.Thread(target=writer) for _ in range(2)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        self.assertEqual(sorted(statuses), [200, 409])

    def test_invalid_payloads(self):
        self.make_user()
        for body in [{"version": "x", "data": self.GOOD}, {"version": -1, "data": self.GOOD}, {"version": 0},
                     {"version": 0, "data": []}, {"version": 0, "data": {"progress": []}}]:
            with self.subTest(body=body):
                self.assertEqual(self.call("PUT", "/api/state", body).status, 400)

    def test_lone_surrogate_key_does_not_crash_saving(self):
        self.make_user()
        raw = b'{"version": 0, "data": {"keys": {"ru": {"\\ud800": {"n": 1}}, "en": {}}}}'
        self.assertEqual(self.call("PUT", "/api/state", raw=raw).status, 200)

    def test_unknown_fields_are_not_stored(self):
        self.make_user()
        self.call("PUT", "/api/state", {"version": 0, "data": {**self.GOOD, "evil": "<script>alert(1)</script>"}})
        self.assertNotIn("evil", self.call("GET", "/api/state").json["data"])

    def test_state_is_isolated_between_users(self):
        self.make_user("a@example.com")
        self.call("PUT", "/api/state", {"version": 0, "data": self.GOOD})
        self.cookie = None
        self.make_user("b@example.com", OTHER_PASSWORD)
        self.assertEqual(self.call("GET", "/api/state").json["data"]["progress"], {})


class SecretsAtRestTests(ApiTestCase):
    def test_database_never_contains_passwords_or_raw_tokens(self):
        self.register()
        verify_token = self.token_from("user@example.com", "verify")
        self.confirm()
        self.login()
        self.call("POST", "/api/forgot", {"email": "user@example.com"})
        reset_token = self.token_from("user@example.com", "reset")
        conn = sqlite3.connect(self.db_path)
        try:
            dump = "\n".join(conn.iterdump())
        finally:
            conn.close()
        for secret in (PASSWORD, verify_token, reset_token, self.cookie):
            self.assertNotIn(secret, dump)


class HttpsCookieTests(ApiTestCase):
    public_url = "https://trainer.example.com/app"

    def test_cookie_is_secure_and_host_prefixed_when_site_is_https(self):
        self.make_user()
        r = self.call("POST", "/api/login", {"email": "user@example.com", "password": PASSWORD}, cookie=None)
        cookie = r.set_cookies[0]
        self.assertTrue(cookie.startswith("__Host-session="))
        for flag in ("Secure", "HttpOnly", "SameSite=Lax", "Path=/"):
            self.assertIn(flag, cookie)
        self.assertNotIn("Domain", cookie)

    def test_links_use_public_url_with_path_not_the_host_header(self):
        self.register("user@example.com")
        r = self.call("POST", "/api/forgot", {"email": "user@example.com"}, headers={"Host": "attacker.example"})
        self.assertEqual(r.status, 200)
        self.assertTrue(all("attacker.example" not in m.body for m in self.mail.sent))
        self.assertIn("https://trainer.example.com/app/#verify=", self.mails_to("user@example.com")[0].body)

    def test_origin_check_matches_normalized_public_url(self):
        self.assertEqual(self.cfg.origin, "https://trainer.example.com")
        r = self.call("POST", "/api/login", {"email": "a@example.com", "password": "x" * 10}, headers={"Origin": "https://trainer.example.com"})
        self.assertEqual(r.status, 401)


class ClientIpTests(ApiTestCase):
    env_extra = {"TRUST_PROXY": "1"}

    def test_trusted_proxy_header_separates_clients(self):
        for i in range(20):
            self.login(f"x{i}@example.com", WRONG, remember=False, headers={"X-Real-IP": "203.0.113.5"})
        blocked = self.login("z@example.com", WRONG, remember=False, headers={"X-Real-IP": "203.0.113.5"})
        other = self.login("z@example.com", WRONG, remember=False, headers={"X-Real-IP": "203.0.113.6"})
        self.assertEqual((blocked.status, other.status), (429, 401))

    def test_attacker_cannot_lock_the_owner_out_from_another_address(self):
        self.make_user()
        self.fail_logins(5, headers={"X-Real-IP": "203.0.113.5"})
        self.assertEqual(self.login(remember=False, headers={"X-Real-IP": "203.0.113.5"}).status, 429)
        self.assertEqual(self.login(remember=False, headers={"X-Real-IP": "198.51.100.7"}).status, 200)

    def test_distributed_guessing_hits_the_account_wide_ceiling(self):
        self.make_user()
        for n in range(20):                                  # 20 адресов по 5 неудач = 100 неудач на один аккаунт за час
            self.fail_logins(5, headers={"X-Real-IP": f"203.0.113.{n + 10}"})
        self.assertEqual(self.login(remember=False, headers={"X-Real-IP": "198.51.100.99"}).status, 429)

    def test_the_account_wide_ceiling_leaves_room_for_the_owner_after_a_slow_attack(self):
        """Атака вполсилы (30 неудач за час с разных адресов) не должна запирать владельца."""
        self.make_user()
        for n in range(6):
            self.fail_logins(5, headers={"X-Real-IP": f"203.0.113.{n + 10}"})
        self.assertEqual(self.login(remember=False, headers={"X-Real-IP": "198.51.100.99"}).status, 200)

    def test_ipv6_clients_of_one_network_share_a_counter(self):
        for i in range(20):
            self.login(f"x{i}@example.com", WRONG, remember=False, headers={"X-Real-IP": f"2001:db8:1:2::{i + 1:x}"})
        r = self.login("z@example.com", WRONG, remember=False, headers={"X-Real-IP": "2001:db8:1:2:ffff:ffff:ffff:fff0"})
        self.assertEqual(r.status, 429)
        other = self.login("z@example.com", WRONG, remember=False, headers={"X-Real-IP": "2001:db8:1:3::1"})
        self.assertEqual(other.status, 401)

    def test_garbage_proxy_header_falls_back_to_peer(self):
        r = self.login("z@example.com", WRONG, remember=False, headers={"X-Real-IP": "not-an-ip"})
        self.assertEqual(r.status, 401)


class UntrustedProxyTests(ApiTestCase):
    def test_header_is_ignored_when_proxy_is_not_trusted(self):
        for i in range(20):
            self.login(f"x{i}@example.com", WRONG, remember=False, headers={"X-Real-IP": f"203.0.113.{i + 10}"})
        r = self.login("z@example.com", WRONG, remember=False, headers={"X-Real-IP": "198.51.100.1"})
        self.assertEqual(r.status, 429)


class IpHelpersTests(unittest.TestCase):
    def test_only_private_and_loopback_peers_may_set_forwarded_headers(self):
        for ok in ["127.0.0.1", "::1", "10.1.2.3", "172.17.0.1", "192.168.65.1", "fd00::1", "::ffff:10.0.0.5"]:
            with self.subTest(ip=ok):
                self.assertTrue(is_trusted_proxy_peer(ok))
        for bad in ["8.8.8.8", "203.0.113.5", "2001:db8::1", "garbage", "", "::ffff:8.8.8.8"]:
            with self.subTest(ip=bad):
                self.assertFalse(is_trusted_proxy_peer(bad))

    def test_ipv6_is_reduced_to_its_64_bit_network(self):
        self.assertEqual(normalize_ip("2001:db8:1:2::1"), normalize_ip("2001:db8:1:2:ffff::9"))
        self.assertNotEqual(normalize_ip("2001:db8:1:2::1"), normalize_ip("2001:db8:1:3::1"))
        self.assertEqual(normalize_ip("203.0.113.5"), "203.0.113.5")
        self.assertEqual(normalize_ip("::ffff:203.0.113.5"), "203.0.113.5")
        self.assertIsNone(normalize_ip("nope"))


if __name__ == "__main__":
    unittest.main()