from __future__ import annotations

import base64
import gzip
import os
import re
import socket
import stat
import subprocess
import sys
import threading
import time
from collections.abc import Generator
from contextlib import contextmanager
from datetime import UTC, datetime, timedelta
from pathlib import Path
from unittest.mock import patch

import pygit2
import pytest
import uvicorn
from fastapi.testclient import TestClient

from pygitweb.auth import ACCESS_TOKEN_COOKIE_NAME
from pygitweb.auth_config import AuthConfig, HttpKey, LocalUser
from pygitweb.config import settings
from pygitweb.conftest import set_client_session
from pygitweb.main import app
from pygitweb.password_hash import hash_password
from pygitweb.permissions import PermissionsMap
from pygitweb.sessions import create_session
from pygitweb.smart_http import (
	_decode_request_body,
	_pkt_flush,
	_pkt_line,
	_service_advertisement,
	append_smart_http_log,
	receive_pack_updated_refs,
	smart_http_log_path,
)

_KEY_ISSUED = datetime(2026, 10, 8, 12, 0, tzinfo=UTC)
_WRITE_KEY = "read-write-key"
_READ_KEY = "read-only-key"
_WRITE_KEY_HASH = hash_password(_WRITE_KEY)
_READ_KEY_HASH = hash_password(_READ_KEY)

_LOCAL_ADMIN = AuthConfig(
	auth_mode="local",
	local_users=[LocalUser(user="alice", password="secret")],
	oauth_permissions=PermissionsMap.default_access(),
)


def _user_with_keys(*, secret: str | None = None, salt: str = "") -> AuthConfig:
	if salt:
		write_secret = secret if secret is not None else hash_password(_WRITE_KEY, config_salt=salt)
		read_secret = hash_password(_READ_KEY, config_salt=salt)
	else:
		write_secret = secret if secret is not None else _WRITE_KEY_HASH
		read_secret = _READ_KEY_HASH
	return AuthConfig(
		auth_mode="local",
		salt=salt,
		local_users=[
			LocalUser(
				user="alice",
				password="secret",
				email="alice@example.com",
				keys=[
					HttpKey(id="laptop", secret=write_secret, readonly=False, time_issued=_KEY_ISSUED),
					HttpKey(
						id="ci",
						secret=read_secret,
						readonly=True,
						time_issued=_KEY_ISSUED,
					),
				],
			)
		],
		oauth_permissions=PermissionsMap.default_access(),
	)


def _push_commands(*refs: str) -> bytes:
	zero = b"0" * 40
	new = b"a" * 40
	body = b""
	for index, ref in enumerate(refs):
		line = zero + b" " + new + b" " + ref.encode("ascii")
		if index == 0:
			line += b"\0report-status"
		body += _pkt_line(line + b"\n")
	return body + _pkt_flush()


@contextmanager
def _auth_enabled(cfg: AuthConfig) -> Generator[None, None, None]:
	with (
		patch.object(settings, "AUTH", True),
		patch("pygitweb.auth_config.auth_config", cfg),
		patch("pygitweb.auth.auth_config", cfg),
		patch("pygitweb.main.auth_config", cfg),
	):
		yield


@pytest.fixture
def local_auth_config() -> Generator[AuthConfig, None, None]:
	with (
		patch("pygitweb.auth_config.auth_config", _LOCAL_ADMIN),
		patch("pygitweb.auth.auth_config", _LOCAL_ADMIN),
		patch("pygitweb.main.auth_config", _LOCAL_ADMIN),
	):
		yield _LOCAL_ADMIN


def _run_git(repo_dir: Path, *args: str, git_config: tuple[str, ...] = ()) -> str:
	completed = subprocess.run(
		[settings.GIT, *git_config, *args],
		cwd=repo_dir,
		capture_output=True,
		text=True,
		check=True,
		env={**os.environ, "GIT_TERMINAL_PROMPT": "0"},
	)
	return completed.stdout.strip()


@contextmanager
def _http_server() -> Generator[int, None, None]:
	with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
		sock.bind(("127.0.0.1", 0))
		port = sock.getsockname()[1]
	config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error", ws="none")
	server = uvicorn.Server(config)
	thread = threading.Thread(target=server.run, daemon=True)
	thread.start()
	deadline = time.time() + 5.0
	while not server.started and time.time() < deadline:
		time.sleep(0.01)
	if not server.started:
		server.should_exit = True
		thread.join(timeout=5.0)
		raise RuntimeError("smart HTTP server did not start")
	try:
		yield port
	finally:
		server.should_exit = True
		thread.join(timeout=5.0)


def _key_git_config() -> tuple[str, ...]:
	helper = f"!f() {{ echo username=alice; echo password={_WRITE_KEY}; }}; f"
	return ("-c", "credential.helper=", "-c", f"credential.helper={helper}")


def _seed_bare_repo(root: Path, bare_dir: Path, message: str = "initial") -> None:
	work_dir = root / "seed-work"
	work_dir.mkdir()
	_run_git(work_dir, "init", "-b", "main")
	_run_git(work_dir, "config", "user.name", "tester")
	_run_git(work_dir, "config", "user.email", "tester@example.com")
	(work_dir / "README.md").write_text(f"{message}\n", encoding="utf-8")
	_run_git(work_dir, "add", "README.md")
	_run_git(work_dir, "commit", "-m", message)
	_run_git(work_dir, "remote", "add", "origin", str(bare_dir))
	_run_git(work_dir, "push", "-u", "origin", "main")


@pytest.fixture(scope="class")
def http_repo_env(tmp_path_factory: pytest.TempPathFactory) -> Generator[dict[str, str], None, None]:
	root = tmp_path_factory.mktemp("smart-http")
	repo_dir = root / "demo"
	log_path = root / "smart_http.log"
	pygit2.init_repository(str(repo_dir), bare=True)
	_seed_bare_repo(root, repo_dir)
	with (
		patch.object(settings, "PROJECTROOT", str(root)),
		patch.object(settings, "PROJECTS_LIST", str(root)),
		patch.object(settings, "PROJECT_MAXDEPTH", 2),
		patch.object(settings, "STRICT_EXPORT", False),
		patch.object(settings, "EXPORT_OK", ""),
		patch.object(settings, "LIST_ALL", True),
		patch.object(settings, "AUTH", False),
		patch("pygitweb.smart_http.smart_http_log_path", return_value=log_path),
	):
		yield {
			"project": "demo",
			"repo_dir": str(repo_dir),
			"root": str(root),
			"log": str(log_path),
		}


class TestSmartHttpHelpers:
	def test_pkt_line_and_flush(self) -> None:
		assert _pkt_line(b"hello\n") == b"000ahello\n"
		assert _pkt_flush() == b"0000"

	def test_receive_pack_updated_refs_stops_at_flush(self) -> None:
		body = _push_commands("refs/heads/main", "refs/tags/v1") + b"PACK leftover"
		assert receive_pack_updated_refs(body) == ["refs/heads/main", "refs/tags/v1"]
		cert = _pkt_line(b"push-cert\0report-status\n") + _pkt_line(b"certificate version 0.1\n")
		assert receive_pack_updated_refs(cert + body) == ["refs/heads/main", "refs/tags/v1"]

	def test_decode_request_body(self) -> None:
		raw = b"0014command=ls-refs\n0000"
		assert _decode_request_body(raw, None) == raw
		assert _decode_request_body(raw, "identity") == raw
		assert _decode_request_body(gzip.compress(raw), "gzip") == raw

	def test_decode_request_body_rejects_invalid_gzip(self) -> None:
		with pytest.raises(Exception) as exc_info:
			_decode_request_body(b"not-gzip", "gzip")
		assert exc_info.value.status_code == 400

	def test_decode_request_body_rejects_unknown_encoding(self) -> None:
		with pytest.raises(Exception) as exc_info:
			_decode_request_body(b"payload", "br")
		assert exc_info.value.status_code == 415

	def test_service_advertisement_matches_git(self, http_repo_env: dict[str, str]) -> None:
		direct = subprocess.run(
			[settings.GIT, "upload-pack", "--advertise-refs", "."],
			cwd=http_repo_env["repo_dir"],
			capture_output=True,
			check=True,
		).stdout
		body = _service_advertisement("git-upload-pack", http_repo_env["project"])
		assert body.startswith(_pkt_line(b"# service=git-upload-pack\n"))
		assert body.endswith(direct)
		assert _pkt_flush() in body


class TestSmartHttpRoutes:
	def test_upload_pack_info_refs(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		resp = client.get(
			f"/http/{http_repo_env['project']}/info/refs",
			params={"service": "git-upload-pack"},
		)
		assert resp.status_code == 200
		assert resp.headers["content-type"].startswith("application/x-git-upload-pack-advertisement")
		assert b"# service=git-upload-pack" in resp.content
		assert b"refs/heads/" in resp.content

	def test_accepts_git_suffix(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		resp = client.get(
			f"/http/{http_repo_env['project']}.git/info/refs",
			params={"service": "git-upload-pack"},
		)
		assert resp.status_code == 200

	def test_unknown_project_404(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		resp = client.get(
			"/http/missing/info/refs",
			params={"service": "git-upload-pack"},
		)
		assert resp.status_code == 404

	def test_upload_pack_accepts_gzip_body(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		request_body = b"0000"
		headers = {
			"Content-Type": "application/x-git-upload-pack-request",
			"Git-Protocol": "version=2",
		}
		plain = client.post(
			f"/http/{http_repo_env['project']}/git-upload-pack",
			content=request_body,
			headers=headers,
		)
		gzipped = client.post(
			f"/http/{http_repo_env['project']}/git-upload-pack",
			content=gzip.compress(request_body),
			headers={**headers, "Content-Encoding": "gzip"},
		)
		assert plain.status_code == gzipped.status_code
		assert plain.content == gzipped.content

	def test_receive_pack_requires_auth_when_enabled(
		self,
		http_repo_env: dict[str, str],
		local_auth_config: AuthConfig,
	) -> None:
		del local_auth_config
		client = TestClient(app)
		with patch.object(settings, "AUTH", True):
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-receive-pack"},
			)
		assert resp.status_code == 401

	def test_receive_pack_allowed_when_authenticated(
		self,
		http_repo_env: dict[str, str],
		local_auth_config: AuthConfig,
	) -> None:
		del local_auth_config
		client = TestClient(app)
		with patch.object(settings, "AUTH", True):
			set_client_session(client, "alice")
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-receive-pack"},
			)
		assert resp.status_code == 200
		assert resp.headers["content-type"].startswith("application/x-git-receive-pack-advertisement")

	def test_fetch_requires_auth_when_enabled(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		client = TestClient(app)
		with _auth_enabled(_user_with_keys()):
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
			)
		assert resp.status_code == 401
		assert "Basic" in resp.headers["www-authenticate"]

	def test_http_key_allows_fetch_and_push(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		client = TestClient(app)
		with _auth_enabled(_user_with_keys()):
			fetch = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", _WRITE_KEY),
			)
			push = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-receive-pack"},
				auth=("Alice@Example.com", _WRITE_KEY),
			)
		assert fetch.status_code == 200
		assert push.status_code == 200

	def test_readonly_key_can_fetch_but_not_push(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		client = TestClient(app)
		with _auth_enabled(_user_with_keys()):
			fetch = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice@example.com", _READ_KEY),
			)
			push = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-receive-pack"},
				auth=("alice", _READ_KEY),
			)
			post_push = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=b"0000",
				auth=("alice", _READ_KEY),
			)
		assert fetch.status_code == 200
		assert push.status_code == 403
		assert post_push.status_code == 403

	def test_login_password_is_not_an_http_key(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		client = TestClient(app)
		with _auth_enabled(_user_with_keys()):
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", "secret"),
			)
			web = client.get("/users/me", auth=("alice", _WRITE_KEY))
		assert resp.status_code == 401
		assert web.status_code == 401

	def test_hashed_http_key_allows_fetch(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		salt = "0123456789abcdef"
		cfg = _user_with_keys(secret=hash_password(_WRITE_KEY, config_salt=salt), salt=salt)
		client = TestClient(app)
		with _auth_enabled(cfg):
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", _WRITE_KEY),
			)
			wrong = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", "not-the-key"),
			)
		assert resp.status_code == 200
		assert wrong.status_code == 401

	def test_http_key_without_project_permission(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		cfg = _user_with_keys()
		cfg = cfg.model_copy(update={"oauth_permissions": PermissionsMap.empty()})
		client = TestClient(app)
		with (
			_auth_enabled(cfg),
			patch("pygitweb.auth.git_get_project_owner", return_value=None),
		):
			fetch = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", _WRITE_KEY),
			)
			push = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-receive-pack"},
				auth=("alice", _WRITE_KEY),
			)
		assert fetch.status_code == 404
		assert push.status_code == 403

	def test_write_ref_whitelist_on_receive_pack(self, http_repo_env: dict[str, str]) -> None:
		restricted = PermissionsMap.model_validate({
			"*": [],
			"pgw.read.*": ["*"],
			"pgw.write.*": ["*"],
			"pgw.write_ref.demo/heads/main": ["other@example.com"],
		})
		allowed = PermissionsMap.model_validate({
			"*": [],
			"pgw.read.*": ["*"],
			"pgw.write.*": ["*"],
			"pgw.write_ref.demo/heads/main": ["alice@example.com"],
			"pgw.write_ref.demo/tags/*": ["alice@example.com"],
		})
		client = TestClient(app)
		denied_body = _push_commands("refs/heads/main")
		with (
			_auth_enabled(_user_with_keys().model_copy(update={"oauth_permissions": restricted})),
			patch("pygitweb.smart_http._run_git_service", return_value=b"") as run,
		):
			denied = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=denied_body,
				auth=("alice", _WRITE_KEY),
			)
			tag = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=_push_commands("refs/tags/v1"),
				auth=("alice", _WRITE_KEY),
			)
		assert denied.status_code == 403
		assert "refs/heads/main" in denied.json()["detail"]
		assert tag.status_code == 403
		assert "refs/tags/v1" in tag.json()["detail"]
		assert run.call_count == 0
		with (
			_auth_enabled(_user_with_keys().model_copy(update={"oauth_permissions": allowed})),
			patch("pygitweb.smart_http._run_git_service", return_value=b"") as run_allowed,
		):
			accepted = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=denied_body,
				auth=("alice", _WRITE_KEY),
			)
			tag_ok = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=_push_commands("refs/tags/v1"),
				auth=("alice", _WRITE_KEY),
			)
		assert accepted.status_code == 200
		assert tag_ok.status_code == 200
		assert run_allowed.call_count == 2
		with (
			_auth_enabled(_user_with_keys().model_copy(update={"oauth_permissions": allowed})),
			patch("pygitweb.smart_http._run_git_service", return_value=b""),
		):
			client.cookies.set(
				ACCESS_TOKEN_COOKIE_NAME,
				create_session("oauthuser", email="alice@example.com", auth_method="oauth"),
			)
			oauth_push = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=denied_body,
			)
		assert oauth_push.status_code == 200

	def test_no_write_ref_falls_back_to_project_write(self, http_repo_env: dict[str, str]) -> None:
		open_write = PermissionsMap.model_validate({
			"*": [],
			"pgw.read.*": ["*"],
			"pgw.write.*": ["*"],
		})
		client = TestClient(app)
		with (
			_auth_enabled(_user_with_keys().model_copy(update={"oauth_permissions": open_write})),
			patch("pygitweb.smart_http._run_git_service", return_value=b"") as run,
		):
			main = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=_push_commands("refs/heads/main"),
				auth=("alice", _WRITE_KEY),
			)
			tag = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=_push_commands("refs/tags/v1"),
				auth=("alice", _WRITE_KEY),
			)
		assert main.status_code == 200
		assert tag.status_code == 200
		assert run.call_count == 2

	def test_ref_push_permission_on_receive_pack(self, http_repo_env: dict[str, str]) -> None:
		restricted = PermissionsMap.model_validate({
			"*": [],
			"pgw.read.*": ["*"],
			"pgw.write.*": ["*"],
			"pgw.write_ref.demo/tags/v*": ["release@example.com"],
			"pgw.write_ref.demo/notes/*": ["release@example.com"],
		})
		client = TestClient(app)
		with (
			_auth_enabled(_user_with_keys().model_copy(update={"oauth_permissions": restricted})),
			patch("pygitweb.smart_http._run_git_service", return_value=b"") as run,
		):
			tag = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=_push_commands("refs/tags/v1"),
				auth=("alice", _WRITE_KEY),
			)
			notes = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=_push_commands("refs/notes/commits"),
				auth=("alice", _WRITE_KEY),
			)
			other_tag = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=_push_commands("refs/tags/nightly"),
				auth=("alice", _WRITE_KEY),
			)
			branch = client.post(
				f"/http/{http_repo_env['project']}/git-receive-pack",
				content=_push_commands("refs/heads/main"),
				auth=("alice", _WRITE_KEY),
			)
		assert tag.status_code == 403
		assert "refs/tags/v1" in tag.json()["detail"]
		assert notes.status_code == 403
		assert other_tag.status_code == 403
		assert branch.status_code == 403
		assert run.call_count == 0

	def test_clone_and_push_over_http(self, http_repo_env: dict[str, str]) -> None:
		with _http_server() as port:
			clone_dir = Path(http_repo_env["root"]) / "clone"
			remote_url = f"http://127.0.0.1:{port}/http/{http_repo_env['project']}.git"
			_run_git(Path(http_repo_env["root"]), "clone", remote_url, str(clone_dir))
			_run_git(clone_dir, "config", "user.name", "tester")
			_run_git(clone_dir, "config", "user.email", "tester@example.com")
			(clone_dir / "new.txt").write_text("pushed\n", encoding="utf-8")
			_run_git(clone_dir, "add", "new.txt")
			_run_git(clone_dir, "commit", "-m", "add new")
			_run_git(clone_dir, "push", "origin", "HEAD")
			repo = pygit2.Repository(http_repo_env["repo_dir"])
			commit = repo.head.peel()
			assert commit.message.strip() == "add new"
			tree = commit.tree
			assert tree["new.txt"].data.decode("utf-8") == "pushed\n"

	def test_expired_http_key_is_rejected_and_unexpired_key_still_works(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		now = datetime.now(UTC)
		cfg = _user_with_keys()
		user = cfg.local_users[0]
		cfg = cfg.model_copy(
			update={
				"local_users": [
					user.model_copy(
						update={
							"keys": [
								HttpKey(
									id="old",
									secret=_WRITE_KEY_HASH,
									readonly=False,
									time_issued=_KEY_ISSUED,
									time_expiry=now - timedelta(seconds=1),
								),
								HttpKey(
									id="ci",
									secret=_READ_KEY_HASH,
									readonly=True,
									time_issued=_KEY_ISSUED,
									time_expiry=now + timedelta(days=1),
								),
							]
						}
					)
				]
			}
		)
		client = TestClient(app)
		with _auth_enabled(cfg):
			expired = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", _WRITE_KEY),
			)
			current = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", _READ_KEY),
			)
		assert expired.status_code == 401
		assert current.status_code == 200

	def test_http_key_rejects_unknown_user_and_user_without_email(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		cfg = AuthConfig(
			auth_mode="local",
			local_users=[
				LocalUser(
					user="alice",
					password="secret",
					email="alice@example.com",
					keys=[HttpKey(id="laptop", secret=_WRITE_KEY_HASH, time_issued=_KEY_ISSUED)],
				),
				LocalUser(
					user="bob",
					password="secret",
					keys=[HttpKey(id="bob-key", secret=hash_password("bob-secret"), time_issued=_KEY_ISSUED)],
				),
			],
			oauth_permissions=PermissionsMap.default_access(),
		)
		client = TestClient(app)
		with _auth_enabled(cfg):
			wrong_user = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("carol", _WRITE_KEY),
			)
			blank_user = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				headers={"Authorization": f"Basic {base64.b64encode(b'   :' + _WRITE_KEY.encode()).decode()}"},
			)
		assert wrong_user.status_code == 401
		assert blank_user.status_code == 401

	def test_malformed_and_non_basic_authorization_rejected(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		client = TestClient(app)
		url = f"/http/{http_repo_env['project']}/info/refs"
		params = {"service": "git-upload-pack"}
		with _auth_enabled(_user_with_keys()):
			not_basic = client.get(url, params=params, headers={"Authorization": "Digest abc"})
			empty_param = client.get(url, params=params, headers={"Authorization": "Basic"})
			bad_b64 = client.get(url, params=params, headers={"Authorization": "Basic @@@@"})
			no_colon = client.get(
				url,
				params=params,
				headers={"Authorization": f"Basic {base64.b64encode(b'alice').decode()}"},
			)
			bad_bearer = client.get(url, params=params, headers={"Authorization": "Bearer not-a-session"})
			client.cookies.set(ACCESS_TOKEN_COOKIE_NAME, "not-a-session")
			bad_cookie = client.get(url, params=params)
			client.cookies.clear()
			bad_utf8 = client.get(
				url,
				params=params,
				headers={"Authorization": f"Basic {base64.b64encode(b'\xff\xfe:secret').decode()}"},
			)
			empty_secret = client.get(
				url,
				params=params,
				headers={"Authorization": f"Basic {base64.b64encode(b'alice:').decode()}"},
			)
			padded_secret = client.get(
				url,
				params=params,
				auth=("  alice  ", f"  {_WRITE_KEY}  "),
			)
		assert not_basic.status_code == 401
		assert empty_param.status_code == 401
		assert bad_b64.status_code == 401
		assert no_colon.status_code == 401
		assert bad_utf8.status_code == 401
		assert empty_secret.status_code == 401
		assert padded_secret.status_code == 200
		assert bad_bearer.status_code == 401
		assert bad_cookie.status_code == 401

	def test_bearer_session_allows_fetch(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		with _auth_enabled(_user_with_keys()):
			token = create_session("alice", email="alice@example.com", auth_method="local")
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				headers={"Authorization": f"Bearer {token}"},
			)
		assert resp.status_code == 200

	def test_keys_only_user_can_fetch_when_local_password_absent(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		cfg = AuthConfig(
			auth_mode="local",
			local_users=[
				LocalUser(
					user="alice",
					email="alice@example.com",
					keys=[HttpKey(id="laptop", secret=_WRITE_KEY_HASH, time_issued=_KEY_ISSUED)],
				)
			],
			oauth_permissions=PermissionsMap.default_access(),
		)
		client = TestClient(app)
		with _auth_enabled(cfg):
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", _WRITE_KEY),
			)
		assert resp.status_code == 200

	def test_unconfigured_auth_without_keys_is_unavailable(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		cfg = AuthConfig(auth_mode="local", local_users=[], oauth_permissions=PermissionsMap.empty())
		client = TestClient(app)
		with _auth_enabled(cfg):
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
			)
		assert resp.status_code == 503

	def test_anonymous_reader_can_fetch_but_not_push(
		self,
		http_repo_env: dict[str, str],
	) -> None:
		cfg = _user_with_keys()
		user = cfg.local_users[0].model_copy(update={"assume_anonymous": True})
		cfg = cfg.model_copy(update={"local_users": [user]})
		client = TestClient(app)
		with _auth_enabled(cfg):
			fetch = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
			)
			push = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-receive-pack"},
			)
		assert fetch.status_code == 200
		assert push.status_code == 401

	def test_clone_and_push_with_http_key(self, http_repo_env: dict[str, str]) -> None:
		with _auth_enabled(_user_with_keys()), _http_server() as port:
			clone_dir = Path(http_repo_env["root"]) / "keyed-clone"
			remote_url = f"http://127.0.0.1:{port}/http/{http_repo_env['project']}.git"
			git_config = _key_git_config()
			_run_git(Path(http_repo_env["root"]), "clone", remote_url, str(clone_dir), git_config=git_config)
			_run_git(clone_dir, "config", "user.name", "tester")
			_run_git(clone_dir, "config", "user.email", "tester@example.com")
			(clone_dir / "from-key.txt").write_text("keyed\n", encoding="utf-8")
			_run_git(clone_dir, "add", "from-key.txt")
			_run_git(clone_dir, "commit", "-m", "add keyed")
			_run_git(clone_dir, "push", "origin", "HEAD", git_config=git_config)
			repo = pygit2.Repository(http_repo_env["repo_dir"])
			commit = repo.head.peel()
			assert commit.message.strip() == "add keyed"
			assert commit.tree["from-key.txt"].data.decode("utf-8") == "keyed\n"


_LOG_LINE = re.compile(r"^(?P<stamp>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z) (?P<rest>.+)$")


def _log_lines(http_repo_env: dict[str, str]) -> list[str]:
	path = Path(http_repo_env["log"])
	if not path.is_file():
		return []
	return path.read_text(encoding="utf-8").splitlines()


def _assert_log_line(line: str, actor: str | None, *, project: str, operation: str) -> None:
	match = _LOG_LINE.fullmatch(line)
	assert match is not None, line
	identity = "" if actor is None else f" {actor}"
	assert match.group("rest") == f"testclient {project} {operation}{identity}"


class TestSmartHttpLogFile:
	def test_default_path_is_under_pygitweb_home(self, tmp_path: Path) -> None:
		with patch("pygitweb.smart_http.Path.home", return_value=tmp_path):
			assert smart_http_log_path() == tmp_path / ".pygitweb" / "smart_http.log"

	def test_append_keeps_earlier_lines_and_restricts_mode(self, tmp_path: Path) -> None:
		path = tmp_path / ".pygitweb" / "smart_http.log"
		with patch("pygitweb.smart_http.smart_http_log_path", return_value=path):
			append_smart_http_log("203.0.113.8", "alice@example.com abcd", project="demo", operation="git-upload-pack")
			append_smart_http_log("203.0.113.9", "INVALID_KEY", project="demo", operation="git-receive-pack")
		lines = path.read_text(encoding="utf-8").splitlines()
		assert lines[0].endswith(" 203.0.113.8 demo git-upload-pack alice@example.com abcd")
		assert lines[1].endswith(" 203.0.113.9 demo git-receive-pack INVALID_KEY")
		assert path.read_bytes().endswith(b"INVALID_KEY\n")
		if sys.platform != "win32":
			assert stat.S_IMODE(path.stat().st_mode) == 0o600


class TestSmartHttpAccessLog:
	def test_each_route_logs_owner_email_and_key_tail(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		project = http_repo_env["project"]
		with _auth_enabled(_user_with_keys()):
			before = len(_log_lines(http_repo_env))
			refs = client.get(
				f"/http/{project}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", _WRITE_KEY),
			)
			upload = client.post(
				f"/http/{project}/git-upload-pack",
				content=b"0000",
				auth=("  alice  ", f"  {_WRITE_KEY}  "),
			)
			push_refs = client.get(
				f"/http/{project}/info/refs",
				params={"service": "git-receive-pack"},
				auth=("Alice@Example.com", _WRITE_KEY),
			)
			receive = client.post(
				f"/http/{project}/git-receive-pack",
				content=b"0000",
				auth=("alice", _WRITE_KEY),
			)
		assert refs.status_code == 200
		assert push_refs.status_code == 200
		assert upload.status_code != 401
		assert receive.status_code != 401
		lines = _log_lines(http_repo_env)[before:]
		assert len(lines) == 4
		operations = (
			"info/refs?service=git-upload-pack",
			"git-upload-pack",
			"info/refs?service=git-receive-pack",
			"git-receive-pack",
		)
		for line, operation in zip(lines, operations, strict=True):
			_assert_log_line(line, "alice@example.com -key", project=project, operation=operation)
			assert _WRITE_KEY not in line

	def test_broken_key_logs_invalid_key_and_expired_key_marks_tail(self, http_repo_env: dict[str, str]) -> None:
		now = datetime.now(UTC)
		cfg = _user_with_keys()
		user = cfg.local_users[0]
		cfg = cfg.model_copy(
			update={
				"local_users": [
					user.model_copy(
						update={
							"keys": [
								HttpKey(
									id="old",
									secret=_WRITE_KEY_HASH,
									readonly=False,
									time_issued=_KEY_ISSUED,
									time_expiry=now - timedelta(seconds=1),
								),
								HttpKey(
									id="ci",
									secret=_READ_KEY_HASH,
									readonly=True,
									time_issued=_KEY_ISSUED,
									time_expiry=now + timedelta(days=1),
								),
							]
						}
					)
				]
			}
		)
		client = TestClient(app)
		project = http_repo_env["project"]
		url = f"/http/{project}/info/refs"
		params = {"service": "git-upload-pack"}
		with _auth_enabled(cfg):
			before = len(_log_lines(http_repo_env))
			expired = client.get(url, params=params, auth=("alice", _WRITE_KEY))
			broken = client.get(url, params=params, auth=("alice", "not-the-key"))
			malformed = client.get(url, params=params, headers={"Authorization": "Basic @@@@"})
			current = client.get(url, params=params, auth=("alice", _READ_KEY))
		assert expired.status_code == 401
		assert broken.status_code == 401
		assert malformed.status_code == 401
		assert current.status_code == 200
		lines = _log_lines(http_repo_env)[before:]
		assert len(lines) == 4
		operation = "info/refs?service=git-upload-pack"
		_assert_log_line(lines[0], "alice@example.com -key (EXPIRED)", project=project, operation=operation)
		_assert_log_line(lines[1], "INVALID_KEY", project=project, operation=operation)
		_assert_log_line(lines[2], "INVALID_KEY", project=project, operation=operation)
		_assert_log_line(lines[3], "alice@example.com -key", project=project, operation=operation)
		logged = "\n".join(lines)
		assert _WRITE_KEY not in logged
		assert _READ_KEY not in logged

	def test_public_fetch_logs_username_instead_of_credentials(self, http_repo_env: dict[str, str]) -> None:
		cfg = _user_with_keys()
		user = cfg.local_users[0].model_copy(update={"assume_anonymous": True})
		cfg = cfg.model_copy(update={"local_users": [user]})
		client = TestClient(app)
		project = http_repo_env["project"]
		with _auth_enabled(cfg):
			before = len(_log_lines(http_repo_env))
			fetch = client.get(
				f"/http/{project}/info/refs",
				params={"service": "git-upload-pack"},
			)
			upload = client.post(f"/http/{project}/git-upload-pack", content=b"0000")
			push = client.get(
				f"/http/{project}/info/refs",
				params={"service": "git-receive-pack"},
			)
			keyed = client.get(
				f"/http/{project}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("alice", _WRITE_KEY),
			)
		assert fetch.status_code == 200
		assert upload.status_code != 401
		assert push.status_code == 401
		assert keyed.status_code == 200
		lines = _log_lines(http_repo_env)[before:]
		assert len(lines) == 4
		_assert_log_line(lines[0], "alice", project=project, operation="info/refs?service=git-upload-pack")
		_assert_log_line(lines[1], "alice", project=project, operation="git-upload-pack")
		_assert_log_line(lines[2], None, project=project, operation="info/refs?service=git-receive-pack")
		_assert_log_line(
			lines[3], "alice@example.com -key", project=project, operation="info/refs?service=git-upload-pack"
		)

	def test_open_fetch_logs_ip_only(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		before = len(_log_lines(http_repo_env))
		resp = client.get(
			f"/http/{http_repo_env['project']}/info/refs",
			params={"service": "git-upload-pack"},
		)
		assert resp.status_code == 200
		lines = _log_lines(http_repo_env)[before:]
		assert len(lines) == 1
		_assert_log_line(
			lines[0], None, project=http_repo_env["project"], operation="info/refs?service=git-upload-pack"
		)

	def test_session_logs_email_without_a_key_tail(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		with _auth_enabled(_user_with_keys()):
			token = create_session("alice", email="alice@example.com", auth_method="local")
			before = len(_log_lines(http_repo_env))
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				headers={"Authorization": f"Bearer {token}"},
			)
		assert resp.status_code == 200
		lines = _log_lines(http_repo_env)[before:]
		assert len(lines) == 1
		_assert_log_line(
			lines[0],
			"alice@example.com",
			project=http_repo_env["project"],
			operation="info/refs?service=git-upload-pack",
		)

	def test_key_owner_without_email_logs_username(self, http_repo_env: dict[str, str]) -> None:
		cfg = AuthConfig(
			auth_mode="local",
			local_users=[
				LocalUser(
					user="bob",
					password="secret",
					keys=[HttpKey(id="bob-key", secret=hash_password("bob-secret"), time_issued=_KEY_ISSUED)],
				)
			],
			oauth_permissions=PermissionsMap.default_access(),
		)
		client = TestClient(app)
		with _auth_enabled(cfg):
			before = len(_log_lines(http_repo_env))
			resp = client.get(
				f"/http/{http_repo_env['project']}/info/refs",
				params={"service": "git-upload-pack"},
				auth=("bob", "bob-secret"),
			)
		assert resp.status_code == 200
		lines = _log_lines(http_repo_env)[before:]
		assert len(lines) == 1
		_assert_log_line(
			lines[0],
			"bob cret",
			project=http_repo_env["project"],
			operation="info/refs?service=git-upload-pack",
		)
		assert "bob-secret" not in lines[0]

	def test_unknown_project_is_not_logged(self, http_repo_env: dict[str, str]) -> None:
		client = TestClient(app)
		before = len(_log_lines(http_repo_env))
		resp = client.get("/http/missing/info/refs", params={"service": "git-upload-pack"})
		assert resp.status_code == 404
		assert _log_lines(http_repo_env)[before:] == []