mirror of
https://github.com/BigBodyCobain/Shadowbroker.git
synced 2026-08-09 12:16:01 +02:00
Ship ShadowBroker v0.9.83 with live Infonet gate messaging and DM protocols.
Gate hashchain replication, Tor/SOCKS transport hardening, terminal session teardown, v0.9.83 UI/changelog, and release digest pins for seamless updater verification.
This commit is contained in:
@@ -51,6 +51,9 @@ class _FakeSocket:
|
||||
def recv(self, _n: int) -> bytes:
|
||||
return self._handshake_response
|
||||
|
||||
def settimeout(self, _timeout: float) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
def __init__(self, *, ok: bool, payload: dict[str, Any], status_code: int = 200) -> None:
|
||||
@@ -76,8 +79,10 @@ def _stub_settings(monkeypatch, *, enabled: bool = True, port: int = 9050) -> No
|
||||
monkeypatch.setattr(
|
||||
"services.config.get_settings", _get_settings, raising=False
|
||||
)
|
||||
# Reset proof cache so each test starts clean.
|
||||
# Reset proof/status cache so each test starts clean.
|
||||
wormhole_supervisor._ARTI_PROOF_CACHE.update({"port": 0, "ok": False, "ts": 0.0})
|
||||
wormhole_supervisor._ARTI_STATUS_CACHE.update({"port": 0, "ready": False, "ts": 0.0})
|
||||
wormhole_supervisor._ARTI_SOCKS_FAILURES = 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Tor hidden service must always publish the mesh SOCKS port."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import socket
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from services import tor_hidden_service as tor_svc
|
||||
|
||||
|
||||
def test_write_torrc_always_includes_socks_port(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(tor_svc, "TOR_DIR", tmp_path)
|
||||
monkeypatch.setattr(tor_svc, "TORRC_PATH", tmp_path / "torrc")
|
||||
monkeypatch.setattr(tor_svc, "TOR_DATA_DIR", tmp_path / "data")
|
||||
|
||||
tor_svc._write_torrc(target_port=8000, socks_port=19050)
|
||||
|
||||
content = tor_svc.TORRC_PATH.read_text(encoding="utf-8")
|
||||
assert "SocksPort 19050" in content
|
||||
assert "HiddenServicePort 8000 127.0.0.1:8000" in content
|
||||
|
||||
|
||||
def test_torrc_has_socks_port_detects_missing_line(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(tor_svc, "TORRC_PATH", tmp_path / "torrc")
|
||||
tor_svc.TORRC_PATH.write_text("HiddenServicePort 8000 127.0.0.1:8000\n", encoding="utf-8")
|
||||
|
||||
assert tor_svc._torrc_has_socks_port(9050) is False
|
||||
|
||||
tor_svc.TORRC_PATH.write_text("SocksPort 9050\n", encoding="utf-8")
|
||||
assert tor_svc._torrc_has_socks_port(9050) is True
|
||||
|
||||
|
||||
def test_local_socks_handshake_ready_accepts_valid_response(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
class FakeSock:
|
||||
def __init__(self) -> None:
|
||||
self._sent = b""
|
||||
|
||||
def settimeout(self, timeout: float) -> None:
|
||||
return None
|
||||
|
||||
def sendall(self, payload: bytes) -> None:
|
||||
self._sent = payload
|
||||
|
||||
def recv(self, size: int) -> bytes:
|
||||
assert self._sent == b"\x05\x01\x00"
|
||||
return b"\x05\x00"
|
||||
|
||||
def __enter__(self) -> "FakeSock":
|
||||
return self
|
||||
|
||||
def __exit__(self, *args: object) -> None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(
|
||||
socket,
|
||||
"create_connection",
|
||||
lambda *_args, **_kwargs: FakeSock(),
|
||||
)
|
||||
assert tor_svc._local_socks_handshake_ready(9050) is True
|
||||
Reference in New Issue
Block a user