test_security_env_injection.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
| 1 | """Phase 2.7 — Environment variable injection security tests. |
| 2 | |
| 3 | Attack surface |
| 4 | -------------- |
| 5 | Muse reads six environment variables: |
| 6 | |
| 7 | MUSE_REPO_ROOT — overrides repository root discovery. |
| 8 | MUSE_AGENT_ID — agent provenance stored in commit records. |
| 9 | MUSE_MODEL_ID — model provenance stored in commit records. |
| 10 | MUSE_TOOLCHAIN_ID — toolchain provenance stored in commit records. |
| 11 | MUSE_PROMPT_HASH — prompt hash stored in commit records. |
| 12 | MUSE_TEST_ENV — passed through to CI gate subprocesses. |
| 13 | |
| 14 | Each variable represents a trust boundary: an attacker who can influence the |
| 15 | process environment (CI pipeline, shared-host user, container escape) can |
| 16 | inject crafted values. |
| 17 | |
| 18 | Attack vectors discovered via muse recon |
| 19 | ----------------------------------------- |
| 20 | 1. **MUSE_REPO_ROOT — empty/whitespace string**: ``pathlib.Path("").resolve()`` |
| 21 | returns the current working directory, so an empty override silently |
| 22 | behaves as if no override was set. Now explicitly ignored (falls through |
| 23 | to directory walk) with a debug log. |
| 24 | |
| 25 | 2. **MUSE_REPO_ROOT — control characters**: a path containing ESC or BEL |
| 26 | could not be a real filesystem path on any OS; now rejected to prevent |
| 27 | logging or display injection of the invalid value. |
| 28 | |
| 29 | 3. **MUSE_REPO_ROOT — overly long path**: values longer than PATH_MAX (4096) |
| 30 | are rejected as injection payloads rather than passed to ``pathlib``. |
| 31 | |
| 32 | 4. **Agent provenance fields (MUSE_AGENT_ID, MUSE_MODEL_ID, |
| 33 | MUSE_TOOLCHAIN_ID, MUSE_PROMPT_HASH)**: the comment in ``commit.py`` |
| 34 | said "prevent control-character-laden strings" but only the length cap |
| 35 | (256 chars) was implemented, not control-character sanitization. |
| 36 | ESC sequences in agent_id → stored in commit records → terminal injection |
| 37 | when provenance is rendered in future display paths, agent dashboards, |
| 38 | or log pipelines. |
| 39 | |
| 40 | 5. **Challenge nonce — CRLF injection**: a nonce containing ``\\r\\n`` would |
| 41 | attempt to inject arbitrary HTTP headers. Python's ``http.client`` blocks |
| 42 | the injection at the wire level (``ValueError: Invalid header value``), but |
| 43 | now rejected at ingestion time so the error is surfaced as a clear |
| 44 | diagnostic rather than a confusing transport exception. |
| 45 | |
| 46 | 6. **Challenge nonce — control characters, excessive length**: non-printable chars |
| 47 | and pathologically long values (> 8192 chars) are now rejected by |
| 48 | ``sanitize_token`` before reaching the HTTP stack. |
| 49 | |
| 50 | Fixes |
| 51 | ----- |
| 52 | - ``sanitize_provenance()`` added to ``muse.core.validation`` — strips all |
| 53 | C0 (0x00–0x1F), DEL (0x7F), and C1 (0x80–0x9F) control characters. |
| 54 | Applied to all four provenance fields in ``muse/cli/commands/commit.py``. |
| 55 | - ``sanitize_token()`` added to ``muse.core.validation`` — strips whitespace, |
| 56 | rejects control chars and values longer than 8192 chars. |
| 57 | - ``find_repo_root()`` in ``muse/core/repo.py`` now explicitly ignores empty |
| 58 | and whitespace-only ``MUSE_REPO_ROOT`` values, logs a debug message, and |
| 59 | rejects values containing control characters or exceeding 4096 chars. |
| 60 | """ |
| 61 | |
| 62 | from __future__ import annotations |
| 63 | |
| 64 | import os |
| 65 | import pathlib |
| 66 | import tempfile |
| 67 | |
| 68 | import pytest |
| 69 | |
| 70 | from muse.core.validation import sanitize_provenance, sanitize_token |
| 71 | from muse.core.repo import find_repo_root |
| 72 | |
| 73 | |
| 74 | # --------------------------------------------------------------------------- |
| 75 | # Helpers |
| 76 | # --------------------------------------------------------------------------- |
| 77 | |
| 78 | def _make_muse_dir(tmp_path: pathlib.Path) -> pathlib.Path: |
| 79 | """Create a minimal .muse/ skeleton and return the repo root.""" |
| 80 | (tmp_path / ".muse").mkdir() |
| 81 | return tmp_path |
| 82 | |
| 83 | |
| 84 | # =========================================================================== |
| 85 | # sanitize_provenance — unit tests |
| 86 | # =========================================================================== |
| 87 | |
| 88 | |
| 89 | class TestSanitizeProvenance: |
| 90 | """sanitize_provenance must strip all C0/DEL/C1 control characters.""" |
| 91 | |
| 92 | def test_clean_string_unchanged(self) -> None: |
| 93 | assert sanitize_provenance("my-agent-v1") == "my-agent-v1" |
| 94 | |
| 95 | def test_empty_string(self) -> None: |
| 96 | assert sanitize_provenance("") == "" |
| 97 | |
| 98 | def test_unicode_allowed(self) -> None: |
| 99 | assert sanitize_provenance("agent-αβγ") == "agent-αβγ" |
| 100 | |
| 101 | def test_spaces_allowed(self) -> None: |
| 102 | """Space (0x20) is not a control char and must be preserved.""" |
| 103 | assert sanitize_provenance("my agent") == "my agent" |
| 104 | |
| 105 | @pytest.mark.parametrize("char,description", [ |
| 106 | ("\x00", "NUL"), |
| 107 | ("\x01", "SOH"), |
| 108 | ("\x07", "BEL"), |
| 109 | ("\x08", "BS"), |
| 110 | ("\x09", "HT (tab)"), |
| 111 | ("\x0a", "LF (newline)"), |
| 112 | ("\x0b", "VT"), |
| 113 | ("\x0c", "FF"), |
| 114 | ("\x0d", "CR"), |
| 115 | ("\x0e", "SO"), |
| 116 | ("\x1b", "ESC — ANSI injection entry point"), |
| 117 | ("\x1f", "US"), |
| 118 | ("\x7f", "DEL"), |
| 119 | ("\x80", "C1 PAD"), |
| 120 | ("\x9b", "CSI — ANSI CSI sequence introducer"), |
| 121 | ("\x9f", "C1 APC"), |
| 122 | ]) |
| 123 | def test_control_char_stripped(self, char: str, description: str) -> None: |
| 124 | result = sanitize_provenance(f"prefix{char}suffix") |
| 125 | assert char not in result |
| 126 | assert "prefixsuffix" == result |
| 127 | |
| 128 | def test_esc_sequence_stripped(self) -> None: |
| 129 | """Full ANSI colour sequence embedded in agent_id must be stripped.""" |
| 130 | result = sanitize_provenance("\x1b[31mmalicious-agent\x1b[0m") |
| 131 | assert "\x1b" not in result |
| 132 | assert result == "[31mmalicious-agent[0m" |
| 133 | |
| 134 | def test_newline_stripped(self) -> None: |
| 135 | """Newline in agent_id would split log lines — must be removed.""" |
| 136 | result = sanitize_provenance("agent\nid\nsplitting") |
| 137 | assert "\n" not in result |
| 138 | assert result == "agentidsplitting" |
| 139 | |
| 140 | def test_crlf_stripped(self) -> None: |
| 141 | result = sanitize_provenance("agent\r\nid") |
| 142 | assert "\r" not in result |
| 143 | assert "\n" not in result |
| 144 | |
| 145 | def test_bel_stripped(self) -> None: |
| 146 | """BEL (0x07) causes terminal bell — must be stripped.""" |
| 147 | result = sanitize_provenance("agent\x07id") |
| 148 | assert "\x07" not in result |
| 149 | |
| 150 | def test_rtl_override_preserved(self) -> None: |
| 151 | """U+202E is not a C0/C1 char; sanitize_provenance does not strip Unicode bidi.""" |
| 152 | # Unicode bidi control characters are a separate concern handled by |
| 153 | # rendering layers. sanitize_provenance only strips C0/DEL/C1. |
| 154 | s = "agent\u202eid" |
| 155 | result = sanitize_provenance(s) |
| 156 | assert result == s |
| 157 | |
| 158 | def test_multiple_control_chars(self) -> None: |
| 159 | payload = "\x1b[31m\x07\x00agent\x1b[0m" |
| 160 | result = sanitize_provenance(payload) |
| 161 | assert "\x1b" not in result |
| 162 | assert "\x07" not in result |
| 163 | assert "\x00" not in result |
| 164 | assert "agent" in result |
| 165 | |
| 166 | def test_does_not_truncate(self) -> None: |
| 167 | """sanitize_provenance does not enforce length — callers do [:256].""" |
| 168 | long_s = "a" * 300 |
| 169 | assert len(sanitize_provenance(long_s)) == 300 |
| 170 | |
| 171 | |
| 172 | # =========================================================================== |
| 173 | # sanitize_token — unit tests |
| 174 | # =========================================================================== |
| 175 | |
| 176 | |
| 177 | class TestSanitizeToken: |
| 178 | """sanitize_token must strip whitespace and reject control chars / overlength.""" |
| 179 | |
| 180 | def test_valid_opaque_token(self) -> None: |
| 181 | tok = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ1c2VyIn0.abc123" |
| 182 | result = sanitize_token(tok) |
| 183 | assert result == tok |
| 184 | |
| 185 | def test_strips_leading_trailing_whitespace(self) -> None: |
| 186 | result = sanitize_token(" mytoken ") |
| 187 | assert result == "mytoken" |
| 188 | |
| 189 | def test_empty_string_returns_none(self) -> None: |
| 190 | assert sanitize_token("") is None |
| 191 | |
| 192 | def test_whitespace_only_returns_none(self) -> None: |
| 193 | assert sanitize_token(" \t\n ") is None |
| 194 | |
| 195 | def test_overlength_returns_none(self) -> None: |
| 196 | assert sanitize_token("a" * 8193) is None |
| 197 | |
| 198 | def test_max_length_accepted(self) -> None: |
| 199 | result = sanitize_token("a" * 8192) |
| 200 | assert result is not None |
| 201 | assert len(result) == 8192 |
| 202 | |
| 203 | @pytest.mark.parametrize("char,description", [ |
| 204 | ("\r", "CR — HTTP header line terminator"), |
| 205 | ("\n", "LF — HTTP header line terminator"), |
| 206 | ("\r\n", "CRLF — HTTP header injection sequence"), |
| 207 | ("\x00", "NUL"), |
| 208 | ("\x01", "SOH"), |
| 209 | ("\x1b", "ESC"), |
| 210 | ("\x1f", "US"), |
| 211 | ("\x7f", "DEL"), |
| 212 | ]) |
| 213 | def test_control_char_returns_none(self, char: str, description: str) -> None: |
| 214 | result = sanitize_token(f"good_token{char}evil") |
| 215 | assert result is None |
| 216 | |
| 217 | def test_crlf_header_injection_blocked(self) -> None: |
| 218 | """Classic HTTP header injection payload must be rejected.""" |
| 219 | payload = "good_token\r\nX-Injected: pwned\r\nAuthorization: MSign attacker" |
| 220 | assert sanitize_token(payload) is None |
| 221 | |
| 222 | def test_unicode_printable_allowed(self) -> None: |
| 223 | """Unicode printable chars (e.g., in opaque tokens) must be accepted.""" |
| 224 | tok = "token-αβγ-δεζ" |
| 225 | result = sanitize_token(tok) |
| 226 | assert result == tok |
| 227 | |
| 228 | def test_bare_api_key_format(self) -> None: |
| 229 | tok = "sk-abc123XYZ_-." |
| 230 | result = sanitize_token(tok) |
| 231 | assert result == tok |
| 232 | |
| 233 | |
| 234 | # =========================================================================== |
| 235 | # find_repo_root — MUSE_REPO_ROOT hardening |
| 236 | # =========================================================================== |
| 237 | |
| 238 | |
| 239 | class TestFindRepoRootEnvHardening: |
| 240 | """find_repo_root must safely handle all MUSE_REPO_ROOT attack payloads.""" |
| 241 | |
| 242 | def _with_env(self, key: str, value: str) -> None: |
| 243 | os.environ[key] = value |
| 244 | |
| 245 | def _clear_env(self, key: str) -> None: |
| 246 | os.environ.pop(key, None) |
| 247 | |
| 248 | def test_empty_string_ignored_falls_through(self, tmp_path: pathlib.Path) -> None: |
| 249 | """Empty MUSE_REPO_ROOT must not redirect to cwd; falls through to walk.""" |
| 250 | _make_muse_dir(tmp_path) |
| 251 | old_cwd = os.getcwd() |
| 252 | try: |
| 253 | os.chdir(tmp_path) |
| 254 | self._with_env("MUSE_REPO_ROOT", "") |
| 255 | result = find_repo_root() |
| 256 | # Should find the cwd repo, not crash or return None |
| 257 | assert result is not None |
| 258 | finally: |
| 259 | self._clear_env("MUSE_REPO_ROOT") |
| 260 | os.chdir(old_cwd) |
| 261 | |
| 262 | def test_whitespace_only_ignored(self, tmp_path: pathlib.Path) -> None: |
| 263 | """Whitespace-only MUSE_REPO_ROOT must be ignored.""" |
| 264 | _make_muse_dir(tmp_path) |
| 265 | old_cwd = os.getcwd() |
| 266 | try: |
| 267 | os.chdir(tmp_path) |
| 268 | self._with_env("MUSE_REPO_ROOT", " \t ") |
| 269 | result = find_repo_root() |
| 270 | # Falls through to walk — finds repo at tmp_path |
| 271 | assert result is not None |
| 272 | finally: |
| 273 | self._clear_env("MUSE_REPO_ROOT") |
| 274 | os.chdir(old_cwd) |
| 275 | |
| 276 | def test_control_char_in_path_returns_none(self) -> None: |
| 277 | """MUSE_REPO_ROOT containing ESC must be rejected, not resolved.""" |
| 278 | self._with_env("MUSE_REPO_ROOT", "/tmp/\x1b[31mattack") |
| 279 | try: |
| 280 | result = find_repo_root() |
| 281 | assert result is None |
| 282 | finally: |
| 283 | self._clear_env("MUSE_REPO_ROOT") |
| 284 | |
| 285 | def test_nul_byte_in_path_returns_none(self) -> None: |
| 286 | """MUSE_REPO_ROOT with embedded NUL (0x01 since 0x00 can't be in env) rejected.""" |
| 287 | self._with_env("MUSE_REPO_ROOT", "/tmp/\x01attack") |
| 288 | try: |
| 289 | result = find_repo_root() |
| 290 | assert result is None |
| 291 | finally: |
| 292 | self._clear_env("MUSE_REPO_ROOT") |
| 293 | |
| 294 | def test_path_max_exceeded_returns_none(self) -> None: |
| 295 | """MUSE_REPO_ROOT longer than PATH_MAX (4096) must be rejected.""" |
| 296 | self._with_env("MUSE_REPO_ROOT", "/tmp/" + "a" * 4092) |
| 297 | try: |
| 298 | result = find_repo_root() |
| 299 | assert result is None |
| 300 | finally: |
| 301 | self._clear_env("MUSE_REPO_ROOT") |
| 302 | |
| 303 | def test_valid_override_to_attacker_dir_with_muse( |
| 304 | self, tmp_path: pathlib.Path |
| 305 | ) -> None: |
| 306 | """MUSE_REPO_ROOT pointing to a dir with real .muse/ is accepted (by design).""" |
| 307 | _make_muse_dir(tmp_path) |
| 308 | self._with_env("MUSE_REPO_ROOT", str(tmp_path)) |
| 309 | try: |
| 310 | result = find_repo_root() |
| 311 | assert result is not None |
| 312 | assert result.resolve() == tmp_path.resolve() |
| 313 | finally: |
| 314 | self._clear_env("MUSE_REPO_ROOT") |
| 315 | |
| 316 | def test_valid_override_to_dir_without_muse_returns_none( |
| 317 | self, tmp_path: pathlib.Path |
| 318 | ) -> None: |
| 319 | """MUSE_REPO_ROOT pointing to a dir without .muse/ returns None.""" |
| 320 | self._with_env("MUSE_REPO_ROOT", str(tmp_path)) |
| 321 | try: |
| 322 | result = find_repo_root() |
| 323 | assert result is None |
| 324 | finally: |
| 325 | self._clear_env("MUSE_REPO_ROOT") |
| 326 | |
| 327 | def test_symlinked_muse_dir_rejected(self, tmp_path: pathlib.Path) -> None: |
| 328 | """MUSE_REPO_ROOT pointing to a dir with a symlinked .muse/ returns None.""" |
| 329 | real = tmp_path / "real" |
| 330 | real.mkdir() |
| 331 | (real / ".muse").mkdir() |
| 332 | attacker = tmp_path / "attacker" |
| 333 | attacker.mkdir() |
| 334 | (attacker / ".muse").symlink_to(real / ".muse") |
| 335 | |
| 336 | self._with_env("MUSE_REPO_ROOT", str(attacker)) |
| 337 | try: |
| 338 | result = find_repo_root() |
| 339 | assert result is None |
| 340 | finally: |
| 341 | self._clear_env("MUSE_REPO_ROOT") |
| 342 | |
| 343 | def test_path_traversal_resolved_safely(self, tmp_path: pathlib.Path) -> None: |
| 344 | """MUSE_REPO_ROOT with ../../ is resolved by pathlib — no .muse/ means None.""" |
| 345 | self._with_env("MUSE_REPO_ROOT", "/tmp/../../nonexistent") |
| 346 | try: |
| 347 | result = find_repo_root() |
| 348 | # Either None (no .muse/ there) or a resolved path without .muse/ → None |
| 349 | assert result is None |
| 350 | finally: |
| 351 | self._clear_env("MUSE_REPO_ROOT") |
| 352 | |
| 353 | def test_nonexistent_path_returns_none(self) -> None: |
| 354 | """MUSE_REPO_ROOT pointing to a non-existent path returns None.""" |
| 355 | self._with_env("MUSE_REPO_ROOT", "/tmp/muse_definitely_does_not_exist_xyz") |
| 356 | try: |
| 357 | result = find_repo_root() |
| 358 | assert result is None |
| 359 | finally: |
| 360 | self._clear_env("MUSE_REPO_ROOT") |
| 361 | |
| 362 | def test_devnull_returns_none(self) -> None: |
| 363 | """/dev/null is not a directory with .muse/ — returns None.""" |
| 364 | self._with_env("MUSE_REPO_ROOT", "/dev/null") |
| 365 | try: |
| 366 | result = find_repo_root() |
| 367 | assert result is None |
| 368 | finally: |
| 369 | self._clear_env("MUSE_REPO_ROOT") |
| 370 | |
| 371 | def test_filesystem_root_returns_none(self) -> None: |
| 372 | """MUSE_REPO_ROOT=/ returns None (no .muse/ at /). Confirms no special behaviour.""" |
| 373 | self._with_env("MUSE_REPO_ROOT", "/") |
| 374 | try: |
| 375 | result = find_repo_root() |
| 376 | assert result is None |
| 377 | finally: |
| 378 | self._clear_env("MUSE_REPO_ROOT") |
| 379 | |
| 380 | |
| 381 | # =========================================================================== |
| 382 | # Agent provenance — end-to-end sanitization in stored records |
| 383 | # =========================================================================== |
| 384 | |
| 385 | |
| 386 | class TestProvenanceSanitizationEndToEnd: |
| 387 | """After commit.py applies sanitize_provenance, commit records must be clean.""" |
| 388 | |
| 389 | def test_esc_in_agent_id_stripped(self) -> None: |
| 390 | """ESC injection in MUSE_AGENT_ID must not survive into the stored value.""" |
| 391 | raw = "\x1b[31mmalicias-agent\x1b[0m" |
| 392 | clean = sanitize_provenance(raw[:256]) |
| 393 | assert "\x1b" not in clean |
| 394 | # Printable text is preserved |
| 395 | assert "malicias-agent" in clean |
| 396 | |
| 397 | def test_newline_in_model_id_stripped(self) -> None: |
| 398 | raw = "gpt-4\nX-Injected: pwned" |
| 399 | clean = sanitize_provenance(raw[:256]) |
| 400 | assert "\n" not in clean |
| 401 | |
| 402 | def test_tab_in_toolchain_id_stripped(self) -> None: |
| 403 | raw = "cursor-agent\tv2" |
| 404 | clean = sanitize_provenance(raw[:256]) |
| 405 | assert "\t" not in clean |
| 406 | |
| 407 | def test_all_c0_chars_stripped_from_prompt_hash(self) -> None: |
| 408 | for byte_val in range(0x00, 0x20): |
| 409 | char = chr(byte_val) |
| 410 | raw = f"hash{char}value" |
| 411 | clean = sanitize_provenance(raw) |
| 412 | assert char not in clean, f"Control char 0x{byte_val:02x} survived sanitize_provenance" |
| 413 | |
| 414 | def test_length_truncation_then_sanitize(self) -> None: |
| 415 | """Simulate commit.py: truncate to _MAX_PROV then sanitize.""" |
| 416 | _MAX_PROV = 256 |
| 417 | payload = "a" * 200 + "\x1b[31m" + "b" * 100 |
| 418 | stored = sanitize_provenance(payload[:_MAX_PROV]) |
| 419 | assert len(stored) <= _MAX_PROV |
| 420 | assert "\x1b" not in stored |
| 421 | |
| 422 | def test_clean_agent_id_survives(self) -> None: |
| 423 | raw = "counterpoint-bot-v2.1" |
| 424 | assert sanitize_provenance(raw[:256]) == raw |
| 425 | |
| 426 | def test_unicode_agent_id_survives(self) -> None: |
| 427 | raw = "agent-αβγ-2024" |
| 428 | assert sanitize_provenance(raw[:256]) == raw |
| 429 | |
| 430 | def test_hyphen_underscore_dot_survive(self) -> None: |
| 431 | raw = "cursor-agent_v2.1" |
| 432 | assert sanitize_provenance(raw[:256]) == raw |
| 433 | |
| 434 | |
| 435 | # =========================================================================== |
| 436 | # sanitize_token — integration with identity.py resolve_token |
| 437 | # =========================================================================== |
| 438 | |
| 439 | |
| 440 | class TestSanitizeTokenIntegration: |
| 441 | """sanitize_token must block CRLF and control chars before HTTP stack.""" |
| 442 | |
| 443 | def test_crlf_blocked_before_http_client(self) -> None: |
| 444 | """A CRLF token must be caught by sanitize_token, not by http.client.""" |
| 445 | payload = "good\r\nX-Injected: pwned" |
| 446 | result = sanitize_token(payload) |
| 447 | assert result is None |
| 448 | |
| 449 | def test_newline_only_blocked(self) -> None: |
| 450 | assert sanitize_token("token\nevil") is None |
| 451 | |
| 452 | def test_cr_only_blocked(self) -> None: |
| 453 | assert sanitize_token("token\revil") is None |
| 454 | |
| 455 | def test_http_client_would_also_block_crlf(self) -> None: |
| 456 | """Demonstrate that Python's http.client blocks CRLF at the wire level. |
| 457 | |
| 458 | This proves the http.client defence exists but our sanitize_token defence |
| 459 | should fire first so the user gets a clear diagnostic. We call |
| 460 | ``putrequest`` + ``putheader`` directly to trigger validation without |
| 461 | opening a socket. |
| 462 | """ |
| 463 | import http.client |
| 464 | |
| 465 | payload = "good_token\r\nX-Injected: pwned" |
| 466 | conn = http.client.HTTPConnection("example.com") |
| 467 | conn.putrequest("GET", "/") |
| 468 | with pytest.raises((ValueError, Exception)): |
| 469 | conn.putheader("Authorization", f"MSign {payload}") |
| 470 | |
| 471 | |
| 472 | # =========================================================================== |
| 473 | # Concurrency — env var reads are snapshot-safe |
| 474 | # =========================================================================== |
| 475 | |
| 476 | |
| 477 | class TestConcurrentEnvVarReads: |
| 478 | """Multiple threads calling sanitize_provenance/sanitize_token concurrently.""" |
| 479 | |
| 480 | def test_concurrent_sanitize_provenance(self) -> None: |
| 481 | import threading |
| 482 | |
| 483 | results: list[str] = [] |
| 484 | errors: list[str] = [] |
| 485 | |
| 486 | def worker(payload: str) -> None: |
| 487 | try: |
| 488 | results.append(sanitize_provenance(payload)) |
| 489 | except Exception as exc: |
| 490 | errors.append(str(exc)) |
| 491 | |
| 492 | payloads = [ |
| 493 | f"agent-{i}\x1b[31m\x07\r\n" for i in range(20) |
| 494 | ] |
| 495 | threads = [threading.Thread(target=worker, args=(p,)) for p in payloads] |
| 496 | for t in threads: |
| 497 | t.start() |
| 498 | for t in threads: |
| 499 | t.join() |
| 500 | |
| 501 | assert errors == [] |
| 502 | assert len(results) == 20 |
| 503 | for r in results: |
| 504 | assert "\x1b" not in r |
| 505 | assert "\x07" not in r |
| 506 | assert "\r" not in r |
| 507 | assert "\n" not in r |
| 508 | |
| 509 | def test_concurrent_sanitize_token(self) -> None: |
| 510 | import threading |
| 511 | |
| 512 | good: list[str] = [] |
| 513 | bad: list[None] = [] |
| 514 | |
| 515 | def worker(tok: str) -> None: |
| 516 | result = sanitize_token(tok) |
| 517 | if result is None: |
| 518 | bad.append(None) |
| 519 | else: |
| 520 | good.append(result) |
| 521 | |
| 522 | valid_tokens = [f"valid-token-{i}" for i in range(10)] |
| 523 | invalid_tokens = [f"bad\r\ntoken-{i}" for i in range(10)] |
| 524 | |
| 525 | threads = [ |
| 526 | threading.Thread(target=worker, args=(t,)) |
| 527 | for t in valid_tokens + invalid_tokens |
| 528 | ] |
| 529 | for t in threads: |
| 530 | t.start() |
| 531 | for t in threads: |
| 532 | t.join() |
| 533 | |
| 534 | assert len(good) == 10 |
| 535 | assert len(bad) == 10 |
| 536 | |
| 537 | |
| 538 | # =========================================================================== |
| 539 | # Fuzzing — random payloads |
| 540 | # =========================================================================== |
| 541 | |
| 542 | |
| 543 | class TestFuzzedEnvVarPayloads: |
| 544 | |
| 545 | @pytest.mark.parametrize("seed", range(20)) |
| 546 | def test_random_control_char_in_provenance_always_stripped(self, seed: int) -> None: |
| 547 | import random |
| 548 | rng = random.Random(seed) |
| 549 | char = chr(rng.randint(0x00, 0x1F)) |
| 550 | payload = f"prefix{char}suffix" |
| 551 | result = sanitize_provenance(payload) |
| 552 | assert char not in result |
| 553 | |
| 554 | @pytest.mark.parametrize("seed", range(10)) |
| 555 | def test_random_crlf_token_always_rejected(self, seed: int) -> None: |
| 556 | import random |
| 557 | rng = random.Random(seed + 50) |
| 558 | crlf = rng.choice(["\r\n", "\r", "\n"]) |
| 559 | payload = f"token{crlf}evil" |
| 560 | assert sanitize_token(payload) is None |
| 561 | |
| 562 | @pytest.mark.parametrize("seed", range(5)) |
| 563 | def test_random_overlength_token_rejected(self, seed: int) -> None: |
| 564 | import random |
| 565 | rng = random.Random(seed + 100) |
| 566 | length = rng.randint(8193, 20000) |
| 567 | payload = "a" * length |
| 568 | assert sanitize_token(payload) is None |
File History
5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e
Merge branch 'dev' into main
Human
29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce
Merge branch 'dev' into main
Human
48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa
feat: Muse — version control for the agent era
Human
101 days ago