diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..a88f3f6 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,63 @@ +name: Test Suite + +on: + push: + branches: [main] + pull_request: + branches: [main] + workflow_dispatch: + +permissions: + contents: read + +jobs: + test: + name: Python ${{ matrix.python-version }} / ${{ matrix.os }} + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + include: + - os: ubuntu-latest + python-version: "3.11" + - os: ubuntu-latest + python-version: "3.12" + - os: windows-latest + python-version: "3.11" + + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Show runtime + run: | + python --version + python -m pip --version + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + python -m pip install -r requirements-dev.txt + + - name: Check dependency consistency + run: python -m pip check + + - name: Compile Python sources + run: python -m compileall -q . + + - name: Run test suite with coverage + run: python -m pytest -q --cov=webapp --cov-report=term-missing --cov-report=xml:coverage.xml --junitxml=test-results.xml + + - name: Upload test artifacts + if: always() + uses: actions/upload-artifact@v4 + with: + name: test-results-${{ matrix.os }}-py${{ matrix.python-version }} + path: | + test-results.xml + coverage.xml diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..ab76891 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,5 @@ +# Development and CI test dependencies +-r requirements.txt + +pytest>=8.0 +pytest-cov>=6.0 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..99a4860 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,21 @@ +"""Shared pytest fixtures for the WebScrapeHelper test suite.""" + +import pytest + +from webapp import create_app + + +@pytest.fixture() +def app(): + application = create_app() + application.config.update( + TESTING=True, + WTF_CSRF_ENABLED=False, + SESSION_SECRET="test-secret", + ) + return application + + +@pytest.fixture() +def client(app): + return app.test_client() diff --git a/tests/test_api_routes.py b/tests/test_api_routes.py new file mode 100644 index 0000000..9d235fe --- /dev/null +++ b/tests/test_api_routes.py @@ -0,0 +1,186 @@ +"""Comprehensive API contract and validation tests.""" + +from types import SimpleNamespace + +import pytest + +import webapp.routes.api as api_module + + +class FakeService: + def __init__(self): + self.analyzer = SimpleNamespace(curve=SimpleNamespace(order=2**256 - 2**32 - 977)) + + def analyze_transaction(self, tx_id): + return {"success": True, "tx_id": tx_id, "weak_signatures": []} + + def analyze_address(self, address, max_txs=500): + return {"success": True, "address": address, "max_txs": max_txs} + + def analyze_ecdsa_pair(self, r1, s1, z1, r2, s2, z2): + return {"success": True, "r1": r1, "s1": s1, "z1": z1, "r2": r2, "s2": s2, "z2": z2} + + def calculate_nonce(self, **values): + return {"success": True, "nonce": "01", "inputs": values} + + def calculate_nonce_from_private_key(self, **values): + return {"success": True, "nonce": "02", "inputs": values} + + def recover_with_known_nonce(self, **values): + return {"success": True, "private_key": "03", "inputs": values} + + def known_addresses(self): + return ["known"] + + def scan_recent_block(self): + return {"success": True, "scanned_transactions": 0} + + def monitor_mempool(self): + return {"success": True, "mempool_scanned": 0} + + +@pytest.fixture() +def fake_service(monkeypatch): + service = FakeService() + monkeypatch.setattr(api_module, "_service", service) + return service + + +def test_health_contract(client): + response = client.get("/api/health") + assert response.status_code == 200 + assert response.get_json() == {"status": "ok", "service": "WebScrapeHelper", "api": "v1"} + + +def test_all_expected_routes_are_registered(app): + routes = {rule.rule for rule in app.url_map.iter_rules()} + expected = { + "/", "/transaction", "/address", "/ecdsa-analysis", + "/api/analyze/transaction", "/api/analyze/address", "/api/analyze/ecdsa", + "/api/calculate/nonce", "/api/calculate/nonce-from-private-key", + "/api/recover/low-s-with-nonce", "/api/recover/malleability-signatures", + "/api/addresses/known", "/api/auto-scan", "/api/monitor-mempool", "/api/health", + } + assert expected <= routes + + +def test_transaction_requires_json_object(client): + response = client.post("/api/analyze/transaction", json=[]) + assert response.status_code == 400 + assert "JSON object body is required" in response.get_json()["error"] + + +def test_transaction_requires_tx_id(client): + response = client.post("/api/analyze/transaction", json={}) + assert response.status_code == 400 + assert "tx_id" in response.get_json()["error"] + + +def test_transaction_rejects_malformed_tx_id(client): + response = client.post("/api/analyze/transaction", json={"tx_id": "not-a-tx"}) + assert response.status_code == 400 + assert "Invalid transaction ID format" in response.get_json()["error"] + + +def test_transaction_delegates_to_service(client, fake_service): + tx_id = "00" * 32 + response = client.post("/api/analyze/transaction", json={"tx_id": tx_id}) + assert response.status_code == 200 + assert response.get_json()["tx_id"] == tx_id + + +def test_address_validation_and_normalization(client, fake_service): + response = client.post( + "/api/analyze/address", + json={"address": " 1BoatSLRHtKNngkdXEeobR76b53LETtpyT ", "max_txs": 10}, + ) + assert response.status_code == 200 + assert response.get_json()["address"] == "1BoatSLRHtKNngkdXEeobR76b53LETtpyT" + assert response.get_json()["max_txs"] == 10 + + +@pytest.mark.parametrize("address", ["", "abc", "0x123", "1invalid0O"]) +def test_address_rejects_invalid_formats(client, address): + response = client.post("/api/analyze/address", json={"address": address}) + assert response.status_code == 400 + + +def test_address_caps_max_txs(client, fake_service): + response = client.post( + "/api/analyze/address", + json={"address": "1BoatSLRHtKNngkdXEeobR76b53LETtpyT", "max_txs": 999999}, + ) + assert response.status_code == 200 + assert response.get_json()["max_txs"] == 5000 + + +@pytest.mark.parametrize("value", ["nope", [], {}, True, None]) +def test_numeric_fields_reject_non_hex_values(client, fake_service, value): + response = client.post( + "/api/calculate/nonce", + json={"r": value, "s1": "01", "s2": "02", "z1": "03", "z2": "04"}, + ) + assert response.status_code == 400 + + +def test_hex_parser_accepts_prefixed_and_unprefixed_values(client, fake_service): + response = client.post( + "/api/calculate/nonce", + json={"r": "0x01", "s1": "02", "s2": "03", "z1": "04", "z2": "05"}, + ) + assert response.status_code == 200 + assert response.get_json()["inputs"] == {"r": 1, "s1": 2, "s2": 3, "z1": 4, "z2": 5} + + +def test_nonce_from_private_key_contract(client, fake_service): + response = client.post( + "/api/calculate/nonce-from-private-key", + json={"r": "01", "s": "02", "z": "03", "x": "04"}, + ) + assert response.status_code == 200 + assert response.get_json()["success"] is True + + +def test_known_nonce_recovery_contract(client, fake_service): + response = client.post( + "/api/recover/low-s-with-nonce", + json={"r": "01", "s": "02", "z": "03", "k": "04"}, + ) + assert response.status_code == 200 + assert response.get_json()["success"] is True + + +def test_method_not_allowed_for_post_only_endpoint(client): + response = client.get("/api/analyze/transaction") + assert response.status_code == 405 + assert response.get_json()["status"] == 405 + + +def test_unknown_api_endpoint_is_json_404(client): + response = client.get("/api/does-not-exist") + assert response.status_code == 404 + assert response.is_json + assert response.get_json()["status"] == 404 + + +def test_known_addresses_endpoint(client, fake_service): + response = client.get("/api/addresses/known") + assert response.status_code == 200 + assert response.get_json() == ["known"] + + +def test_legacy_transaction_alias_uses_same_contract(client, fake_service): + response = client.post("/api/analyze_transaction", json={"tx_id": "00" * 32}) + assert response.status_code == 200 + assert response.get_json()["success"] is True + + +def test_unexpected_service_error_becomes_500(client, monkeypatch): + class BrokenService: + def analyze_transaction(self, tx_id): + raise RuntimeError("boom") + + monkeypatch.setattr(api_module, "_service", BrokenService()) + response = client.post("/api/analyze/transaction", json={"tx_id": "00" * 32}) + assert response.status_code == 500 + assert response.get_json() == {"error": "Internal server error", "status": 500} diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..e16f258 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,46 @@ +"""Application-factory, error-handler, and security-header tests.""" + + +def test_security_headers_are_present(client): + response = client.get("/api/health") + assert response.headers["X-Content-Type-Options"] == "nosniff" + assert response.headers["X-Frame-Options"] == "SAMEORIGIN" + assert response.headers["Referrer-Policy"] == "strict-origin-when-cross-origin" + + +def test_static_resources_receive_cache_header(client): + response = client.get("/static/js/api-client.js") + assert response.status_code == 200 + assert "max-age=3600" in response.headers["Cache-Control"] + + +def test_html_404_is_not_json(client): + response = client.get("/route-that-does-not-exist") + assert response.status_code == 404 + assert response.is_json is False + assert response.get_data(as_text=True) == "Not Found" + + +def test_api_404_is_json(client): + response = client.get("/api/route-that-does-not-exist") + assert response.status_code == 404 + assert response.get_json() == {"error": "Endpoint not found", "status": 404} + + +def test_api_405_is_json(client): + response = client.get("/api/analyze/address") + assert response.status_code == 405 + assert response.get_json() == {"error": "Method not allowed", "status": 405} + + +def test_html_routes_render(client): + for path in ("/", "/transaction", "/address", "/ecdsa-analysis"): + response = client.get(path) + assert response.status_code == 200, path + assert "text/html" in response.content_type + + +def test_app_uses_test_configuration(app): + assert app.config["TESTING"] is True + assert app.config["MAX_CONTENT_LENGTH"] > 0 + assert app.config["ANALYSIS_MAX_TXS"] > 0 diff --git a/tests/test_bitcoin_service.py b/tests/test_bitcoin_service.py new file mode 100644 index 0000000..bba3300 --- /dev/null +++ b/tests/test_bitcoin_service.py @@ -0,0 +1,103 @@ +"""Deterministic unit tests for the Bitcoin/ECDSA service layer.""" + +from ecdsa import curves + +from webapp.services.bitcoin import BitcoinService + + +N = curves.SECP256k1.order +G = curves.SECP256k1.generator + + +def make_signature(private_key, nonce, z): + r = (nonce * G).x() % N + s = ((z + r * private_key) * pow(nonce, -1, N)) % N + return r, s + + +def test_nonce_recovery_formula_round_trip(): + private_key = 0x123456789ABCDEF + nonce = 0x23456789ABCDEF1 + z1 = 0x111111111111111111 + z2 = 0x222222222222222222 + r1, s1 = make_signature(private_key, nonce, z1) + r2, s2 = make_signature(private_key, nonce, z2) + + service = BitcoinService() + result = service.calculate_nonce(r1, s1, s2, z1, z2) + + assert result["success"] is True + assert int(result["nonce"], 16) == nonce + assert r1 == r2 + + +def test_ecdsa_pair_recovery_round_trip(): + private_key = 0x3456789ABCDEF123 + nonce = 0x456789ABCDEF1234 + z1 = 0x1010101010101010 + z2 = 0x2020202020202020 + r1, s1 = make_signature(private_key, nonce, z1) + r2, s2 = make_signature(private_key, nonce, z2) + + service = BitcoinService() + result = service.analyze_ecdsa_pair(r1, s1, z1, r2, s2, z2) + + assert result["success"] is True + assert int(result["k"], 16) == nonce + assert int(result["x"], 16) == private_key + + +def test_nonce_from_known_private_key_round_trip(): + private_key = 0x5566778899AABBCC + nonce = 0x1122334455667788 + z = 0x9999999999999999 + r, s = make_signature(private_key, nonce, z) + + service = BitcoinService() + result = service.calculate_nonce_from_private_key(r, s, z, private_key) + + assert result["success"] is True + assert int(result["nonce"], 16) == nonce + + +def test_private_key_recovery_from_known_nonce_round_trip(): + private_key = 0x123456789ABC1234 + nonce = 0xABCD123456789 + z = 0x13579BDF + r, s = make_signature(private_key, nonce, z) + + service = BitcoinService() + result = service.recover_with_known_nonce(r, s, z, nonce) + + assert result["success"] is True + assert int(result["private_key"], 16) == private_key + + +def test_nonce_recovery_rejects_identical_s_values(): + service = BitcoinService() + try: + service.calculate_nonce(1, 2, 2, 3, 4) + except ValueError as exc: + assert "identical" in str(exc) + else: + raise AssertionError("Expected ValueError") + + +def test_pair_recovery_rejects_different_r_values(): + service = BitcoinService() + try: + service.analyze_ecdsa_pair(1, 2, 3, 4, 5, 6) + except ValueError as exc: + assert "R values must match" in str(exc) + else: + raise AssertionError("Expected ValueError") + + +def test_known_nonce_recovery_rejects_zero_nonce(): + service = BitcoinService() + try: + service.recover_with_known_nonce(1, 2, 3, 0) + except ValueError as exc: + assert "scalar range" in str(exc) + else: + raise AssertionError("Expected ValueError") diff --git a/tests/test_frontend_contract.py b/tests/test_frontend_contract.py new file mode 100644 index 0000000..c53f10d --- /dev/null +++ b/tests/test_frontend_contract.py @@ -0,0 +1,49 @@ +"""Static frontend integration/contract tests.""" + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def read(path): + return (ROOT / path).read_text(encoding="utf-8") + + +def test_frontend_controller_assets_exist(): + assert (ROOT / "static/js/api-client.js").is_file() + assert (ROOT / "static/js/main-controller.js").is_file() + assert (ROOT / "static/js/ecdsa_analyzer.js").is_file() + + +def test_base_template_loads_single_application_controller(): + base = read("base.html") + assert "static/js/api-client.js" in base + assert "static/js/main-controller.js" in base + assert "static/js/main.js" not in base + + +def test_application_frontend_uses_central_api_client(): + controller = read("static/js/main-controller.js") + analyzer = read("static/js/ecdsa_analyzer.js") + client = read("static/js/api-client.js") + + assert "apiClient" in controller + assert "apiClient" in analyzer + assert "fetch(" in client + assert "fetch(" not in controller + assert "fetch(" not in analyzer + + +def test_api_client_exposes_expected_backend_operations(): + client = read("static/js/api-client.js") + for operation in ( + "analyzeTransaction", + "analyzeAddress", + "analyzeECDSA", + "calculateNonce", + "calculateNonceFromPrivateKey", + "recoverWithKnownNonce", + "analyzeMalleability", + ): + assert operation in client diff --git a/tests/test_malleability_api.py b/tests/test_malleability_api.py new file mode 100644 index 0000000..da551ab --- /dev/null +++ b/tests/test_malleability_api.py @@ -0,0 +1,47 @@ +"""Tests for the corrected ECDSA signature-malleability endpoint.""" + +from webapp.services.bitcoin import BitcoinService + + +N = BitcoinService().analyzer.curve.order + + +def test_malleability_detects_s_and_n_minus_s(client): + s = 123456789 + complement = N - s + response = client.post( + "/api/recover/malleability-signatures", + json={"r": "01", "z": "02", "s_values": [format(s, "x"), format(complement, "x")]}, + ) + + assert response.status_code == 200 + data = response.get_json() + assert data["success"] is True + assert data["malleable"] is True + assert data["private_key_recovered"] is False + assert len(data["malleable_pairs"]) == 1 + + +def test_malleability_does_not_flag_unrelated_s_values(client): + response = client.post( + "/api/recover/malleability-signatures", + json={"r": "01", "z": "02", "s_values": ["03", "04"]}, + ) + assert response.status_code == 200 + assert response.get_json()["malleable"] is False + + +def test_malleability_requires_two_s_values(client): + response = client.post( + "/api/recover/malleability-signatures", + json={"r": "01", "z": "02", "s_values": ["03"]}, + ) + assert response.status_code == 400 + + +def test_malleability_rejects_out_of_range_r(client): + response = client.post( + "/api/recover/malleability-signatures", + json={"r": format(N, "x"), "z": "02", "s_values": ["03", "04"]}, + ) + assert response.status_code == 400 diff --git a/tests/test_service_network.py b/tests/test_service_network.py new file mode 100644 index 0000000..6acd50e --- /dev/null +++ b/tests/test_service_network.py @@ -0,0 +1,70 @@ +"""Unit tests for external-network service behavior using mocked HTTP responses.""" + +import requests + +import webapp.routes.api as api_module +from webapp.services.bitcoin import BitcoinService + + +class FakeResponse: + def __init__(self, payload): + self.payload = payload + + def raise_for_status(self): + return None + + def json(self): + return self.payload + + +def test_scan_recent_block_uses_latest_hash_and_limit(monkeypatch): + calls = [] + + def fake_get(url, timeout): + calls.append((url, timeout)) + if url.endswith("latestblock"): + return FakeResponse({"hash": "block-hash"}) + return FakeResponse({"tx": [{"hash": "tx1"}, {"hash": "tx2"}, {"hash": "tx3"}]}) + + service = BitcoinService() + monkeypatch.setattr("webapp.services.bitcoin.requests.get", fake_get) + monkeypatch.setattr(service, "analyze_transaction", lambda tx_id: {"private_keys_found": 0}) + + result = service.scan_recent_block(limit=2) + + assert result["success"] is True + assert result["scanned_transactions"] == 2 + assert result["block_hash"] == "block-hash" + assert calls == [ + ("https://blockchain.info/latestblock", 10), + ("https://blockchain.info/rawblock/block-hash", 10), + ] + + +def test_monitor_mempool_uses_limit(monkeypatch): + def fake_get(url, timeout): + assert url == "https://blockchain.info/unconfirmed-transactions?format=json" + assert timeout == 10 + return FakeResponse({"txs": [{"hash": "tx1"}, {"hash": "tx2"}]}) + + service = BitcoinService() + monkeypatch.setattr("webapp.services.bitcoin.requests.get", fake_get) + monkeypatch.setattr(service, "analyze_transaction", lambda tx_id: {"private_keys_found": 0}) + + result = service.monitor_mempool(limit=1) + + assert result["success"] is True + assert result["mempool_scanned"] == 1 + assert result["vulnerable_transactions"] == 0 + + +def test_network_error_is_mapped_to_502(client, monkeypatch): + class BrokenService: + def analyze_transaction(self, tx_id): + raise requests.RequestException("upstream down") + + monkeypatch.setattr(api_module, "_service", BrokenService()) + response = client.post("/api/analyze/transaction", json={"tx_id": "00" * 32}) + + assert response.status_code == 502 + assert response.get_json() == {"error": "Upstream Bitcoin service unavailable", "status": 502}