Spaces:
Sleeping
Sleeping
| """Tests for the titan security modules.""" | |
| from __future__ import annotations | |
| import pytest | |
| from titan import ( | |
| FAISS_PARTITION_SEED, | |
| HardwareHandshake, | |
| DeviceAuthError, | |
| PacketValidationError, | |
| SignalNoiseFilter, | |
| ) | |
| from titan.device_auth import HWID_ENV_VAR | |
| class TestConstants: | |
| def test_faiss_partition_seed_value(self): | |
| assert FAISS_PARTITION_SEED == 9293 | |
| def test_faiss_partition_seed_is_int(self): | |
| assert isinstance(FAISS_PARTITION_SEED, int) | |
| class TestSignalNoiseFilter: | |
| def _good_packet(self, payload: str = "ok") -> dict: | |
| return {"device_id": "d1", "timestamp": 1, "payload": payload} | |
| def test_accepts_well_formed_packet(self): | |
| f = SignalNoiseFilter() | |
| assert f.is_valid(self._good_packet()) is True | |
| def test_rejects_non_mapping(self): | |
| f = SignalNoiseFilter() | |
| assert f.is_valid("not a dict") is False | |
| assert f.is_valid(None) is False | |
| def test_rejects_missing_field(self): | |
| f = SignalNoiseFilter() | |
| bad = self._good_packet() | |
| del bad["device_id"] | |
| assert f.is_valid(bad) is False | |
| def test_rejects_empty_field(self): | |
| f = SignalNoiseFilter() | |
| bad = self._good_packet() | |
| bad["device_id"] = "" | |
| assert f.is_valid(bad) is False | |
| def test_rejects_oversized_payload(self): | |
| f = SignalNoiseFilter(max_packet_bytes=4) | |
| assert f.is_valid(self._good_packet("12345")) is False | |
| def test_validate_raises_on_invalid(self): | |
| f = SignalNoiseFilter() | |
| with pytest.raises(PacketValidationError): | |
| f.validate({"device_id": "d"}) | |
| def test_filter_stream_drops_and_counts(self): | |
| f = SignalNoiseFilter() | |
| packets = [ | |
| self._good_packet("a"), | |
| {"device_id": "d"}, # missing fields | |
| self._good_packet("b"), | |
| "garbage", # not a mapping | |
| ] | |
| kept = list(f.filter_stream(packets)) | |
| assert len(kept) == 2 | |
| assert f.accepted_count == 2 | |
| assert f.dropped_count == 2 | |
| def test_rejects_non_positive_max(self): | |
| with pytest.raises(ValueError): | |
| SignalNoiseFilter(max_packet_bytes=0) | |
| class TestHardwareHandshake: | |
| def test_requires_hwid(self, monkeypatch): | |
| monkeypatch.delenv(HWID_ENV_VAR, raising=False) | |
| with pytest.raises(DeviceAuthError): | |
| HardwareHandshake() | |
| def test_reads_hwid_from_env(self, monkeypatch): | |
| monkeypatch.setenv(HWID_ENV_VAR, "env-hwid") | |
| hs = HardwareHandshake() | |
| challenge = hs.generate_challenge() | |
| assert hs.verify(challenge, hs.sign(challenge)) | |
| def test_sign_and_verify_roundtrip(self): | |
| hs = HardwareHandshake(hwid="secret-hwid") | |
| challenge = hs.generate_challenge(16) | |
| sig = hs.sign(challenge) | |
| assert isinstance(sig, str) | |
| assert hs.verify(challenge, sig) is True | |
| def test_verify_rejects_bad_signature(self): | |
| hs = HardwareHandshake(hwid="secret-hwid") | |
| challenge = hs.generate_challenge() | |
| assert hs.verify(challenge, "deadbeef") is False | |
| def test_verify_rejects_tampered_challenge(self): | |
| hs = HardwareHandshake(hwid="secret-hwid") | |
| challenge = hs.generate_challenge() | |
| sig = hs.sign(challenge) | |
| assert hs.verify(challenge + b"x", sig) is False | |
| def test_different_hwids_produce_different_signatures(self): | |
| a = HardwareHandshake(hwid="hwid-a") | |
| b = HardwareHandshake(hwid="hwid-b") | |
| challenge = b"same-challenge" | |
| assert a.sign(challenge) != b.sign(challenge) | |
| assert a.verify(challenge, b.sign(challenge)) is False | |
| def test_rejects_unsupported_algorithm(self): | |
| with pytest.raises(DeviceAuthError): | |
| HardwareHandshake(hwid="x", algorithm="definitely-not-real") | |
| def test_generate_challenge_length(self): | |
| hs = HardwareHandshake(hwid="x") | |
| assert len(hs.generate_challenge(24)) == 24 | |
| def test_generate_challenge_rejects_non_positive(self): | |
| hs = HardwareHandshake(hwid="x") | |
| with pytest.raises(ValueError): | |
| hs.generate_challenge(0) | |
| def test_sign_rejects_non_bytes(self): | |
| hs = HardwareHandshake(hwid="x") | |
| with pytest.raises(TypeError): | |
| hs.sign("not-bytes") # type: ignore[arg-type] | |
| def test_verify_rejects_non_string_signature(self): | |
| hs = HardwareHandshake(hwid="x") | |
| assert hs.verify(b"chal", 12345) is False # type: ignore[arg-type] | |