microsoft/hve-core
Publicmirrored from https://github.com/microsoft/hve-coreAvailable
.github/skills/experimental/mural/tests/fuzz_harness.py
768lines · modecode
| 1 | # Copyright (c) Microsoft Corporation. |
| 2 | # SPDX-License-Identifier: MIT |
| 3 | """Polyglot fuzz harness for Mural skill helper logic. |
| 4 | |
| 5 | Runs as a pytest test when Atheris is not installed. |
| 6 | Runs as an Atheris coverage-guided fuzz target when executed directly. |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | import os |
| 12 | import sys |
| 13 | import tempfile |
| 14 | from contextlib import suppress |
| 15 | |
| 16 | import mural |
| 17 | import pytest |
| 18 | |
| 19 | # The Atheris/libFuzzer subprocess can run with ``HOME`` unset, which makes |
| 20 | # bare ``~`` expansion fall back to ``pathlib.Path.home()``. Provide a |
| 21 | # deterministic fallback so the home-directory branch is exercised instead of |
| 22 | # aborting the run. Note: ``~user`` forms (e.g. ``~unknownuser``) still raise |
| 23 | # ``RuntimeError`` from ``expanduser`` regardless of ``HOME``; targets that feed |
| 24 | # arbitrary paths to ``expanduser`` suppress that explicitly. |
| 25 | os.environ.setdefault("HOME", tempfile.gettempdir()) |
| 26 | |
| 27 | try: |
| 28 | import atheris |
| 29 | except ImportError: |
| 30 | atheris = None |
| 31 | FUZZING = False |
| 32 | else: |
| 33 | FUZZING = True |
| 34 | |
| 35 | |
| 36 | class _FakeTokenResponse: |
| 37 | """Minimal HTTP response stand-in for ``_parse_token_response`` fuzzing.""" |
| 38 | |
| 39 | def __init__(self, status: int, content_type: str, body: bytes) -> None: |
| 40 | self.status = status |
| 41 | self.headers = {"Content-Type": content_type} |
| 42 | self._body = body |
| 43 | |
| 44 | def read(self, n: int = -1) -> bytes: |
| 45 | if n is None or n < 0: |
| 46 | data, self._body = self._body, b"" |
| 47 | return data |
| 48 | chunk, self._body = self._body[:n], self._body[n:] |
| 49 | return chunk |
| 50 | |
| 51 | |
| 52 | def fuzz_redact(data: bytes) -> None: |
| 53 | """Fuzz the secret-redaction helper with arbitrary text.""" |
| 54 | provider = atheris.FuzzedDataProvider(data) |
| 55 | text = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 56 | mural._redact(text) |
| 57 | |
| 58 | |
| 59 | def fuzz_validate_mural_id(data: bytes) -> None: |
| 60 | """Fuzz mural-id validation; only ``MuralValidationError`` is expected.""" |
| 61 | provider = atheris.FuzzedDataProvider(data) |
| 62 | candidate = provider.ConsumeUnicodeNoSurrogates(60) |
| 63 | with suppress(mural.MuralValidationError): |
| 64 | mural._validate_mural_id(candidate) |
| 65 | |
| 66 | |
| 67 | def fuzz_extract_field(data: bytes) -> None: |
| 68 | """Fuzz nested field extraction across representative payload shapes.""" |
| 69 | provider = atheris.FuzzedDataProvider(data) |
| 70 | payload = { |
| 71 | "id": provider.ConsumeUnicodeNoSurrogates(20), |
| 72 | "fields": { |
| 73 | "title": provider.ConsumeUnicodeNoSurrogates(40), |
| 74 | "labels": [provider.ConsumeUnicodeNoSurrogates(12) for _ in range(3)], |
| 75 | "metadata": {"count": provider.ConsumeIntInRange(0, 50)}, |
| 76 | }, |
| 77 | } |
| 78 | path_options = [ |
| 79 | "id", |
| 80 | "fields.title", |
| 81 | "fields.labels.0", |
| 82 | "fields.metadata.count", |
| 83 | provider.ConsumeUnicodeNoSurrogates(30), |
| 84 | ] |
| 85 | mural._extract_field( |
| 86 | payload, |
| 87 | path_options[provider.ConsumeIntInRange(0, len(path_options) - 1)], |
| 88 | ) |
| 89 | |
| 90 | |
| 91 | def fuzz_parse_pagination_cursor(data: bytes) -> None: |
| 92 | """Fuzz opaque cursor decoding; only ``MuralValidationError`` is expected.""" |
| 93 | provider = atheris.FuzzedDataProvider(data) |
| 94 | token = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 95 | with suppress(mural.MuralValidationError): |
| 96 | mural._parse_pagination_cursor(token) |
| 97 | |
| 98 | |
| 99 | def fuzz_validate_asset_url(data: bytes) -> None: |
| 100 | """Fuzz the SSRF allowlist; only ``MuralSecurityError`` is expected.""" |
| 101 | provider = atheris.FuzzedDataProvider(data) |
| 102 | url = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 103 | with suppress(mural.MuralSecurityError): |
| 104 | mural._validate_asset_url(url) |
| 105 | |
| 106 | |
| 107 | def fuzz_validate_redirect_uri(data: bytes) -> None: |
| 108 | """Fuzz the OAuth loopback redirect URI validator. |
| 109 | |
| 110 | Only ``MuralSecurityError`` is expected. |
| 111 | """ |
| 112 | provider = atheris.FuzzedDataProvider(data) |
| 113 | uri = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 114 | with suppress(mural.MuralSecurityError): |
| 115 | mural._validate_redirect_uri(uri) |
| 116 | |
| 117 | |
| 118 | def fuzz_parse_json_arg(data: bytes) -> None: |
| 119 | """Fuzz JSON argument parsing; only ``MuralValidationError`` is expected.""" |
| 120 | provider = atheris.FuzzedDataProvider(data) |
| 121 | text = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 122 | with suppress(mural.MuralValidationError): |
| 123 | mural._parse_json_arg(text, "--body") |
| 124 | |
| 125 | |
| 126 | def fuzz_verify_pkce(data: bytes) -> None: |
| 127 | """Fuzz PKCE verification with arbitrary verifier/challenge pairs.""" |
| 128 | provider = atheris.FuzzedDataProvider(data) |
| 129 | verifier = provider.ConsumeUnicodeNoSurrogates(64) |
| 130 | challenge = provider.ConsumeUnicodeNoSurrogates(64) |
| 131 | mural._verify_pkce(verifier, challenge) |
| 132 | |
| 133 | |
| 134 | def fuzz_extract_error_payload(data: bytes) -> None: |
| 135 | """Fuzz Mural API error payload extraction with arbitrary body bytes and headers.""" |
| 136 | provider = atheris.FuzzedDataProvider(data) |
| 137 | body_len = provider.ConsumeIntInRange(0, max(0, provider.remaining_bytes() - 1)) |
| 138 | body = provider.ConsumeBytes(body_len) |
| 139 | header_choice = provider.ConsumeIntInRange(0, 3) |
| 140 | headers_obj: object | None |
| 141 | if header_choice == 0: |
| 142 | headers_obj = None |
| 143 | elif header_choice == 1: |
| 144 | headers_obj = {} |
| 145 | elif header_choice == 2: |
| 146 | headers_obj = {"X-Request-Id": provider.ConsumeUnicodeNoSurrogates(32)} |
| 147 | else: |
| 148 | headers_obj = {"x-request-id": provider.ConsumeUnicodeNoSurrogates(32)} |
| 149 | mural._extract_error_payload(body, headers_obj) |
| 150 | |
| 151 | |
| 152 | def fuzz_build_authorize_url(data: bytes) -> None: |
| 153 | """Fuzz OAuth authorize URL builder; only ``MuralError`` is expected.""" |
| 154 | provider = atheris.FuzzedDataProvider(data) |
| 155 | client_id = provider.ConsumeUnicodeNoSurrogates(32) |
| 156 | redirect_uri = provider.ConsumeUnicodeNoSurrogates(64) |
| 157 | state = provider.ConsumeUnicodeNoSurrogates(32) |
| 158 | code_challenge = provider.ConsumeUnicodeNoSurrogates(64) |
| 159 | scopes = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 160 | with suppress(mural.MuralError): |
| 161 | mural._build_authorize_url( |
| 162 | client_id, redirect_uri, state, code_challenge, scopes |
| 163 | ) |
| 164 | |
| 165 | |
| 166 | def fuzz_loopback_callback_request(data: bytes) -> None: |
| 167 | """Fuzz ``_LoopbackHandler`` request parsing with arbitrary raw HTTP bytes. |
| 168 | |
| 169 | Drives the handler via an in-memory ``socket.socketpair`` so the full |
| 170 | request-line, header, path, and query-string parsing path runs against |
| 171 | fuzzer-controlled bytes. Any exception inside the handler is suppressed |
| 172 | because crash-free behavior on malformed input is the property under test. |
| 173 | """ |
| 174 | import http.server |
| 175 | import socket |
| 176 | import threading |
| 177 | |
| 178 | # Cap input size so the in-process socket buffer never blocks. |
| 179 | payload = data[:8192] |
| 180 | if not payload: |
| 181 | return |
| 182 | |
| 183 | class _FuzzServer(http.server.HTTPServer): |
| 184 | """Minimal HTTPServer stand-in that records callback results.""" |
| 185 | |
| 186 | # Skip socket setup; we plumb the request socket manually. |
| 187 | def __init__(self) -> None: # noqa: D401 - stub |
| 188 | self.server_port = 8765 |
| 189 | self.server_address = ("127.0.0.1", 8765) |
| 190 | self.callback_result = mural._CallbackResult() |
| 191 | self.callback_received = threading.Event() |
| 192 | |
| 193 | def shutdown_request(self, request: object) -> None: # noqa: D401 - stub |
| 194 | with suppress(Exception): |
| 195 | request.close() # type: ignore[union-attr] |
| 196 | |
| 197 | def close_request(self, request: object) -> None: # noqa: D401 - stub |
| 198 | with suppress(Exception): |
| 199 | request.close() # type: ignore[union-attr] |
| 200 | |
| 201 | server_sock, client_sock = socket.socketpair() |
| 202 | try: |
| 203 | # Push fuzzer bytes into the handler's read side, then half-close |
| 204 | # so the parser sees EOF rather than blocking. |
| 205 | with suppress(OSError): |
| 206 | server_sock.sendall(payload) |
| 207 | with suppress(OSError): |
| 208 | server_sock.shutdown(socket.SHUT_WR) |
| 209 | with suppress(Exception): |
| 210 | mural._LoopbackHandler(client_sock, ("127.0.0.1", 0), _FuzzServer()) |
| 211 | finally: |
| 212 | with suppress(OSError): |
| 213 | server_sock.close() |
| 214 | with suppress(OSError): |
| 215 | client_sock.close() |
| 216 | |
| 217 | |
| 218 | def fuzz_parse_token_response(data: bytes) -> None: |
| 219 | """Fuzz the OAuth token endpoint response parser. |
| 220 | |
| 221 | Only ``MuralError`` (covering ``MuralAPIError`` and ``ResponseTooLarge``) |
| 222 | is expected. |
| 223 | """ |
| 224 | provider = atheris.FuzzedDataProvider(data) |
| 225 | ct_choice = provider.ConsumeIntInRange(0, 4) |
| 226 | if ct_choice == 0: |
| 227 | content_type = "application/json" |
| 228 | elif ct_choice == 1: |
| 229 | content_type = "application/json; charset=utf-8" |
| 230 | elif ct_choice == 2: |
| 231 | content_type = "text/html" |
| 232 | elif ct_choice == 3: |
| 233 | content_type = "" |
| 234 | else: |
| 235 | content_type = provider.ConsumeUnicodeNoSurrogates(32) |
| 236 | status = provider.ConsumeIntInRange(100, 599) |
| 237 | body = provider.ConsumeBytes(provider.remaining_bytes()) |
| 238 | resp = _FakeTokenResponse(status, content_type, body) |
| 239 | with suppress(mural.MuralError): |
| 240 | mural._parse_token_response(resp) |
| 241 | |
| 242 | |
| 243 | def fuzz_unwrap_value_envelope(data: bytes) -> None: |
| 244 | """Fuzz the single-GET envelope unwrap helper. |
| 245 | |
| 246 | The function never raises; this asserts crash-free behavior and the |
| 247 | documented passthrough invariants across representative input shapes. |
| 248 | """ |
| 249 | provider = atheris.FuzzedDataProvider(data) |
| 250 | shape = provider.ConsumeIntInRange(0, 7) |
| 251 | inner = provider.ConsumeUnicodeNoSurrogates(32) |
| 252 | record: object |
| 253 | if shape == 0: |
| 254 | record = {"value": {"id": inner}} |
| 255 | elif shape == 1: |
| 256 | record = {"value": inner} |
| 257 | elif shape == 2: |
| 258 | record = {"value": [inner]} |
| 259 | elif shape == 3: |
| 260 | record = {"value": {"id": inner}, "next": inner} |
| 261 | elif shape == 4: |
| 262 | record = {"id": inner} |
| 263 | elif shape == 5: |
| 264 | record = [inner] |
| 265 | elif shape == 6: |
| 266 | record = inner |
| 267 | else: |
| 268 | record = None |
| 269 | result = mural._unwrap_value_envelope(record) |
| 270 | if ( |
| 271 | isinstance(record, dict) |
| 272 | and list(record.keys()) == ["value"] |
| 273 | and isinstance(record["value"], dict) |
| 274 | ): |
| 275 | assert result is record["value"] |
| 276 | else: |
| 277 | assert result is record |
| 278 | |
| 279 | |
| 280 | def fuzz_validate_hyperlink(data: bytes) -> None: |
| 281 | """Fuzz the hyperlink validator; only ``MuralValidationError`` is expected.""" |
| 282 | provider = atheris.FuzzedDataProvider(data) |
| 283 | value = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 284 | with suppress(mural.MuralValidationError): |
| 285 | mural._validate_hyperlink(value) |
| 286 | |
| 287 | |
| 288 | def fuzz_validate_tag_text(data: bytes) -> None: |
| 289 | """Fuzz the tag text validator; only ``MuralValidationError`` is expected.""" |
| 290 | provider = atheris.FuzzedDataProvider(data) |
| 291 | value = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 292 | with suppress(mural.MuralValidationError): |
| 293 | mural._validate_tag_text(value) |
| 294 | |
| 295 | |
| 296 | def fuzz_validate_area_layout(data: bytes) -> None: |
| 297 | """Fuzz the area layout validator; only ``MuralValidationError`` is expected.""" |
| 298 | provider = atheris.FuzzedDataProvider(data) |
| 299 | choice = provider.ConsumeIntInRange(0, 4) |
| 300 | if choice == 0: |
| 301 | value: object = "free" |
| 302 | elif choice == 1: |
| 303 | value = "column" |
| 304 | elif choice == 2: |
| 305 | value = "row" |
| 306 | elif choice == 3: |
| 307 | value = "" |
| 308 | else: |
| 309 | value = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 310 | with suppress(mural.MuralValidationError): |
| 311 | mural._validate_area_layout(value) |
| 312 | |
| 313 | |
| 314 | def fuzz_validate_profile_name(data: bytes) -> None: |
| 315 | """Fuzz the profile-name validator; only ``MuralValidationError`` is expected.""" |
| 316 | provider = atheris.FuzzedDataProvider(data) |
| 317 | candidate = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 318 | with suppress(mural.MuralValidationError): |
| 319 | mural._validate_profile_name(candidate) |
| 320 | |
| 321 | |
| 322 | def fuzz_validate_profile(data: bytes) -> None: |
| 323 | """Fuzz the persisted-profile shape validator across representative dicts. |
| 324 | |
| 325 | ``_validate_profile`` raises bare :class:`mural.MuralError` on shape or |
| 326 | type violations, so suppression is on the parent exception class. |
| 327 | """ |
| 328 | provider = atheris.FuzzedDataProvider(data) |
| 329 | shape = provider.ConsumeIntInRange(0, 6) |
| 330 | valid_base: dict[str, object] = { |
| 331 | "client_id": provider.ConsumeUnicodeNoSurrogates(32), |
| 332 | "access_token": provider.ConsumeUnicodeNoSurrogates(64), |
| 333 | "token_type": "Bearer", |
| 334 | "obtained_at": provider.ConsumeIntInRange(0, 2_000_000_000), |
| 335 | "expires_at": provider.ConsumeIntInRange(0, 2_000_000_000), |
| 336 | } |
| 337 | profile: object |
| 338 | if shape == 0: |
| 339 | profile = valid_base |
| 340 | elif shape == 1: |
| 341 | profile = {**valid_base, "expires_at": True} |
| 342 | elif shape == 2: |
| 343 | profile = {**valid_base, "expires_at": "0"} |
| 344 | elif shape == 3: |
| 345 | partial = dict(valid_base) |
| 346 | partial.pop("client_id", None) |
| 347 | profile = partial |
| 348 | elif shape == 4: |
| 349 | profile = {} |
| 350 | elif shape == 5: |
| 351 | profile = None |
| 352 | else: |
| 353 | profile = [valid_base] |
| 354 | with suppress(mural.MuralError): |
| 355 | mural._validate_profile(profile) |
| 356 | |
| 357 | |
| 358 | def fuzz_parse_rate_limit_headers(data: bytes) -> None: |
| 359 | """Fuzz ``X-RateLimit-*`` parsing against an isolated bucket. |
| 360 | |
| 361 | The parser does not raise on malformed values; this asserts the dict |
| 362 | contract and crash-free behavior. An isolated :class:`mural._TokenBucket` |
| 363 | keeps the module-level limiter unaffected. |
| 364 | """ |
| 365 | provider = atheris.FuzzedDataProvider(data) |
| 366 | remaining_choice = provider.ConsumeIntInRange(0, 4) |
| 367 | if remaining_choice == 0: |
| 368 | remaining_value: object = provider.ConsumeUnicodeNoSurrogates(16) |
| 369 | elif remaining_choice == 1: |
| 370 | remaining_value = str(provider.ConsumeIntInRange(-100, 1000)) |
| 371 | elif remaining_choice == 2: |
| 372 | remaining_value = "0" |
| 373 | elif remaining_choice == 3: |
| 374 | remaining_value = "" |
| 375 | else: |
| 376 | remaining_value = None |
| 377 | reset_choice = provider.ConsumeIntInRange(0, 3) |
| 378 | if reset_choice == 0: |
| 379 | reset_value: object = provider.ConsumeUnicodeNoSurrogates(16) |
| 380 | elif reset_choice == 1: |
| 381 | reset_value = str(provider.ConsumeIntInRange(0, 1_000_000)) |
| 382 | elif reset_choice == 2: |
| 383 | reset_value = "" |
| 384 | else: |
| 385 | reset_value = None |
| 386 | headers: dict[str, object] = {} |
| 387 | if remaining_value is not None: |
| 388 | headers["X-RateLimit-Remaining"] = remaining_value |
| 389 | if reset_value is not None: |
| 390 | headers["X-RateLimit-Reset"] = reset_value |
| 391 | bucket = mural._TokenBucket() |
| 392 | result = mural._parse_rate_limit_headers(headers, bucket=bucket) |
| 393 | assert isinstance(result, dict) |
| 394 | assert set(result.keys()) == {"remaining", "reset"} |
| 395 | |
| 396 | |
| 397 | def fuzz_profile_from_credential_path(data: bytes) -> None: |
| 398 | """Fuzz credential-path → profile-name parsing; never raises.""" |
| 399 | import pathlib as _pathlib |
| 400 | |
| 401 | provider = atheris.FuzzedDataProvider(data) |
| 402 | name = provider.ConsumeUnicodeNoSurrogates(provider.remaining_bytes()) |
| 403 | # ``pathlib.Path`` rejects embedded NULs; strip so the parser is the focus. |
| 404 | name = name.replace("\x00", "").replace("/", "") |
| 405 | if not name: |
| 406 | return |
| 407 | path = _pathlib.Path("/tmp") / name |
| 408 | result = mural._profile_from_credential_path(path) |
| 409 | assert isinstance(result, str) |
| 410 | |
| 411 | |
| 412 | def fuzz_resolve_credential_file(data: bytes) -> None: |
| 413 | """Fuzz credential-file path resolution from arbitrary environ + profile.""" |
| 414 | import pathlib as _pathlib |
| 415 | |
| 416 | provider = atheris.FuzzedDataProvider(data) |
| 417 | profile = provider.ConsumeUnicodeNoSurrogates(32).replace("\x00", "") |
| 418 | if not profile: |
| 419 | profile = mural.DEFAULT_PROFILE_NAME |
| 420 | explicit = provider.ConsumeUnicodeNoSurrogates(64).replace("\x00", "") |
| 421 | xdg = provider.ConsumeUnicodeNoSurrogates(64).replace("\x00", "") |
| 422 | appdata = provider.ConsumeUnicodeNoSurrogates(64).replace("\x00", "") |
| 423 | environ: dict[str, str] = {} |
| 424 | shape = provider.ConsumeIntInRange(0, 3) |
| 425 | if shape == 0 and explicit: |
| 426 | environ[mural.ENV_ENV_FILE] = explicit |
| 427 | elif shape == 1 and xdg: |
| 428 | environ[mural.ENV_XDG_CONFIG_HOME] = xdg |
| 429 | elif shape == 2 and appdata: |
| 430 | environ["APPDATA"] = appdata |
| 431 | # ``expanduser`` raises RuntimeError for unresolvable ``~user`` forms in the |
| 432 | # fuzzed MURAL_ENV_FILE value; that is a valid input outcome, not a defect. |
| 433 | with suppress(ValueError, RuntimeError): |
| 434 | result = mural._resolve_credential_file(profile, environ) |
| 435 | assert isinstance(result, _pathlib.Path) |
| 436 | |
| 437 | |
| 438 | FUZZ_TARGETS = [ |
| 439 | fuzz_redact, |
| 440 | fuzz_validate_mural_id, |
| 441 | fuzz_extract_field, |
| 442 | fuzz_parse_pagination_cursor, |
| 443 | fuzz_validate_asset_url, |
| 444 | fuzz_validate_redirect_uri, |
| 445 | fuzz_parse_json_arg, |
| 446 | fuzz_verify_pkce, |
| 447 | fuzz_extract_error_payload, |
| 448 | fuzz_build_authorize_url, |
| 449 | fuzz_loopback_callback_request, |
| 450 | fuzz_parse_token_response, |
| 451 | fuzz_unwrap_value_envelope, |
| 452 | fuzz_validate_hyperlink, |
| 453 | fuzz_validate_tag_text, |
| 454 | fuzz_validate_area_layout, |
| 455 | fuzz_validate_profile_name, |
| 456 | fuzz_validate_profile, |
| 457 | fuzz_parse_rate_limit_headers, |
| 458 | fuzz_profile_from_credential_path, |
| 459 | fuzz_resolve_credential_file, |
| 460 | ] |
| 461 | |
| 462 | |
| 463 | def fuzz_dispatch(data: bytes) -> None: |
| 464 | """Route input to one fuzz target.""" |
| 465 | if len(data) < 2: |
| 466 | return |
| 467 | target_index = data[0] % len(FUZZ_TARGETS) |
| 468 | FUZZ_TARGETS[target_index](data[1:]) |
| 469 | |
| 470 | |
| 471 | class TestMuralFuzzHarness: |
| 472 | """Property tests mirroring fuzz-target behavior.""" |
| 473 | |
| 474 | @pytest.mark.parametrize( |
| 475 | ("text", "should_change"), |
| 476 | [ |
| 477 | ("plain log line with no secrets", False), |
| 478 | ("Authorization: Bearer abc.def.ghi token=value", True), |
| 479 | ("", False), |
| 480 | ], |
| 481 | ) |
| 482 | def test_redact_is_safe_for_arbitrary_text( |
| 483 | self, text: str, should_change: bool |
| 484 | ) -> None: |
| 485 | result = mural._redact(text) |
| 486 | assert isinstance(result, str) |
| 487 | if should_change: |
| 488 | assert result != text or text == "" |
| 489 | |
| 490 | @pytest.mark.parametrize( |
| 491 | "candidate", |
| 492 | ["workspace1.mural-abc123", "ws.mural-xyz"], |
| 493 | ) |
| 494 | def test_validate_mural_id_accepts_valid_values(self, candidate: str) -> None: |
| 495 | assert mural._validate_mural_id(candidate) == candidate |
| 496 | |
| 497 | @pytest.mark.parametrize( |
| 498 | "candidate", |
| 499 | ["", "../etc/passwd", "ws/mural", "ws\\mural", "ws.mural\x00", "no-dot"], |
| 500 | ) |
| 501 | def test_validate_mural_id_rejects_invalid_values(self, candidate: str) -> None: |
| 502 | with pytest.raises(mural.MuralValidationError): |
| 503 | mural._validate_mural_id(candidate) |
| 504 | |
| 505 | def test_extract_field_handles_nested_values(self) -> None: |
| 506 | payload = { |
| 507 | "fields": { |
| 508 | "title": "Sticky note", |
| 509 | "labels": ["a", "b", "c"], |
| 510 | "metadata": {"count": 3}, |
| 511 | } |
| 512 | } |
| 513 | assert mural._extract_field(payload, "fields.title") == "Sticky note" |
| 514 | assert mural._extract_field(payload, "fields.labels.1") == "b" |
| 515 | assert mural._extract_field(payload, "fields.metadata.count") == 3 |
| 516 | assert mural._extract_field(payload, "fields.missing") is None |
| 517 | assert mural._extract_field(payload, "") == payload |
| 518 | |
| 519 | def test_parse_pagination_cursor_round_trip(self) -> None: |
| 520 | import base64 |
| 521 | import json as _json |
| 522 | |
| 523 | token = ( |
| 524 | base64.urlsafe_b64encode(_json.dumps({"page": 2}).encode("utf-8")) |
| 525 | .rstrip(b"=") |
| 526 | .decode("ascii") |
| 527 | ) |
| 528 | assert mural._parse_pagination_cursor(token) == {"page": 2} |
| 529 | |
| 530 | @pytest.mark.parametrize( |
| 531 | "token", |
| 532 | ["", "!!!not-base64!!!", "Zm9vYmFy"], |
| 533 | ) |
| 534 | def test_parse_pagination_cursor_rejects_invalid(self, token: str) -> None: |
| 535 | with pytest.raises(mural.MuralValidationError): |
| 536 | mural._parse_pagination_cursor(token) |
| 537 | |
| 538 | @pytest.mark.parametrize( |
| 539 | "url", |
| 540 | [ |
| 541 | "", |
| 542 | "http://account.blob.core.windows.net/upload", |
| 543 | "https://example.com/upload", |
| 544 | "https://user:pass@account.blob.core.windows.net/upload", |
| 545 | "https://account.blob.core.windows.net/upload#frag", |
| 546 | "https://10.0.0.1/upload", |
| 547 | ], |
| 548 | ) |
| 549 | def test_validate_asset_url_rejects_unsafe(self, url: str) -> None: |
| 550 | with pytest.raises(mural.MuralSecurityError): |
| 551 | mural._validate_asset_url(url) |
| 552 | |
| 553 | def test_validate_asset_url_accepts_azure_blob(self) -> None: |
| 554 | mural._validate_asset_url( |
| 555 | "https://account.blob.core.windows.net/c/asset?sig=xyz" |
| 556 | ) |
| 557 | |
| 558 | def test_parse_json_arg_round_trip(self) -> None: |
| 559 | assert mural._parse_json_arg('{"x":1}', "--body") == {"x": 1} |
| 560 | |
| 561 | def test_parse_json_arg_rejects_invalid(self) -> None: |
| 562 | with pytest.raises(mural.MuralValidationError): |
| 563 | mural._parse_json_arg("not json", "--body") |
| 564 | |
| 565 | def test_verify_pkce_round_trip(self) -> None: |
| 566 | verifier, challenge = mural._generate_pkce_pair() |
| 567 | assert mural._verify_pkce(verifier, challenge) is True |
| 568 | assert mural._verify_pkce(verifier, "wrong") is False |
| 569 | |
| 570 | def test_parse_token_response_round_trip(self) -> None: |
| 571 | resp = _FakeTokenResponse(200, "application/json", b'{"access_token":"x"}') |
| 572 | assert mural._parse_token_response(resp) == {"access_token": "x"} |
| 573 | |
| 574 | @pytest.mark.parametrize( |
| 575 | ("content_type", "body"), |
| 576 | [ |
| 577 | ("text/html", b"<html/>"), |
| 578 | ("", b"{}"), |
| 579 | ("application/json", b"not json"), |
| 580 | ("application/json", b"[1,2,3]"), |
| 581 | ("application/json", b"null"), |
| 582 | ], |
| 583 | ) |
| 584 | def test_parse_token_response_rejects_invalid( |
| 585 | self, content_type: str, body: bytes |
| 586 | ) -> None: |
| 587 | resp = _FakeTokenResponse(200, content_type, body) |
| 588 | with pytest.raises(mural.MuralAPIError): |
| 589 | mural._parse_token_response(resp) |
| 590 | |
| 591 | def test_validate_hyperlink_accepts_short_string(self) -> None: |
| 592 | assert mural._validate_hyperlink("https://example.com") == "https://example.com" |
| 593 | |
| 594 | @pytest.mark.parametrize( |
| 595 | "value", |
| 596 | [ |
| 597 | None, |
| 598 | "", |
| 599 | 123, |
| 600 | "x" * (mural._MAX_HYPERLINK_LEN + 1), |
| 601 | "javascript:alert(1)", |
| 602 | "data:text/html,x", |
| 603 | "vbscript:msg", |
| 604 | "file:///etc/passwd", |
| 605 | ], |
| 606 | ) |
| 607 | def test_validate_hyperlink_rejects_invalid(self, value: object) -> None: |
| 608 | with pytest.raises(mural.MuralValidationError): |
| 609 | mural._validate_hyperlink(value) |
| 610 | |
| 611 | def test_validate_tag_text_accepts_short_string(self) -> None: |
| 612 | assert mural._validate_tag_text("todo") == "todo" |
| 613 | |
| 614 | @pytest.mark.parametrize( |
| 615 | "value", |
| 616 | [None, "", 7, "x" * (mural._MAX_TAG_TEXT_LEN + 1)], |
| 617 | ) |
| 618 | def test_validate_tag_text_rejects_invalid(self, value: object) -> None: |
| 619 | with pytest.raises(mural.MuralValidationError): |
| 620 | mural._validate_tag_text(value) |
| 621 | |
| 622 | @pytest.mark.parametrize("value", ["free", "column", "row"]) |
| 623 | def test_validate_area_layout_accepts_valid(self, value: str) -> None: |
| 624 | assert mural._validate_area_layout(value) == value |
| 625 | |
| 626 | @pytest.mark.parametrize("value", ["", "grid", "FREE", None, 1]) |
| 627 | def test_validate_area_layout_rejects_invalid(self, value: object) -> None: |
| 628 | with pytest.raises(mural.MuralValidationError): |
| 629 | mural._validate_area_layout(value) |
| 630 | |
| 631 | @pytest.mark.parametrize( |
| 632 | "name", |
| 633 | ["default", "prod", "user_1", "team.dev", "test-env", "_underscore"], |
| 634 | ) |
| 635 | def test_validate_profile_name_accepts_valid(self, name: str) -> None: |
| 636 | assert mural._validate_profile_name(name) == name |
| 637 | |
| 638 | @pytest.mark.parametrize( |
| 639 | "name", |
| 640 | [None, "", 7, "..", "../etc", "has space", ".leading-dot", "x" * 33], |
| 641 | ) |
| 642 | def test_validate_profile_name_rejects_invalid(self, name: object) -> None: |
| 643 | with pytest.raises(mural.MuralValidationError): |
| 644 | mural._validate_profile_name(name) |
| 645 | |
| 646 | def test_validate_profile_accepts_minimal_valid(self) -> None: |
| 647 | profile = { |
| 648 | "client_id": "abc", |
| 649 | "access_token": "tok", |
| 650 | "token_type": "Bearer", |
| 651 | "obtained_at": 0, |
| 652 | "expires_at": 0, |
| 653 | } |
| 654 | assert mural._validate_profile(profile) is None |
| 655 | |
| 656 | @pytest.mark.parametrize( |
| 657 | "profile", |
| 658 | [ |
| 659 | None, |
| 660 | [], |
| 661 | "string", |
| 662 | {}, |
| 663 | { |
| 664 | "client_id": "c", |
| 665 | "access_token": "t", |
| 666 | "token_type": "B", |
| 667 | "obtained_at": 0, |
| 668 | "expires_at": True, |
| 669 | }, |
| 670 | { |
| 671 | "client_id": "c", |
| 672 | "access_token": "t", |
| 673 | "token_type": "B", |
| 674 | "obtained_at": 0, |
| 675 | "expires_at": "0", |
| 676 | }, |
| 677 | ], |
| 678 | ) |
| 679 | def test_validate_profile_rejects_invalid(self, profile: object) -> None: |
| 680 | with pytest.raises(mural.MuralError): |
| 681 | mural._validate_profile(profile) |
| 682 | |
| 683 | def test_parse_rate_limit_headers_parses_ints(self) -> None: |
| 684 | bucket = mural._TokenBucket() |
| 685 | headers = {"X-RateLimit-Remaining": "5", "X-RateLimit-Reset": "60"} |
| 686 | assert mural._parse_rate_limit_headers(headers, bucket=bucket) == { |
| 687 | "remaining": 5, |
| 688 | "reset": 60, |
| 689 | } |
| 690 | |
| 691 | def test_parse_rate_limit_headers_returns_none_for_missing(self) -> None: |
| 692 | bucket = mural._TokenBucket() |
| 693 | assert mural._parse_rate_limit_headers({}, bucket=bucket) == { |
| 694 | "remaining": None, |
| 695 | "reset": None, |
| 696 | } |
| 697 | |
| 698 | def test_parse_rate_limit_headers_drains_bucket_on_exhaustion(self) -> None: |
| 699 | bucket = mural._TokenBucket() |
| 700 | bucket.tokens = 4.0 |
| 701 | headers = {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": "30"} |
| 702 | mural._parse_rate_limit_headers(headers, bucket=bucket) |
| 703 | assert bucket.tokens == 0.0 |
| 704 | |
| 705 | def test_profile_from_credential_path_extracts_profile(self) -> None: |
| 706 | import pathlib as _pathlib |
| 707 | |
| 708 | assert ( |
| 709 | mural._profile_from_credential_path(_pathlib.Path("/tmp/mural.team_x.env")) |
| 710 | == "team_x" |
| 711 | ) |
| 712 | |
| 713 | def test_profile_from_credential_path_falls_back_to_default(self) -> None: |
| 714 | import pathlib as _pathlib |
| 715 | |
| 716 | assert ( |
| 717 | mural._profile_from_credential_path(_pathlib.Path("/tmp/random.txt")) |
| 718 | == mural.DEFAULT_PROFILE_NAME |
| 719 | ) |
| 720 | |
| 721 | def test_resolve_credential_file_honors_explicit_env(self) -> None: |
| 722 | path = mural._resolve_credential_file( |
| 723 | "default", {mural.ENV_ENV_FILE: "/tmp/explicit.env"} |
| 724 | ) |
| 725 | assert str(path) == "/tmp/explicit.env" |
| 726 | |
| 727 | def test_resolve_credential_file_uses_xdg_when_set(self) -> None: |
| 728 | path = mural._resolve_credential_file( |
| 729 | "team_a", {mural.ENV_XDG_CONFIG_HOME: "/tmp/xdg"} |
| 730 | ) |
| 731 | assert str(path) == "/tmp/xdg/hve-core/mural.team_a.env" |
| 732 | |
| 733 | |
| 734 | _CORPUS_ROOT = __import__("pathlib").Path(__file__).parent / "corpus" |
| 735 | |
| 736 | |
| 737 | def _collect_corpus_seeds() -> list[tuple[str, str]]: |
| 738 | if not _CORPUS_ROOT.is_dir(): |
| 739 | return [] |
| 740 | seeds: list[tuple[str, str]] = [] |
| 741 | for target in FUZZ_TARGETS: |
| 742 | target_dir = _CORPUS_ROOT / target.__name__ |
| 743 | if not target_dir.is_dir(): |
| 744 | continue |
| 745 | for seed_path in sorted(target_dir.iterdir()): |
| 746 | if seed_path.is_file() and seed_path.suffix == ".bin": |
| 747 | seeds.append((target.__name__, str(seed_path))) |
| 748 | return seeds |
| 749 | |
| 750 | |
| 751 | _CORPUS_SEEDS = _collect_corpus_seeds() |
| 752 | |
| 753 | |
| 754 | @pytest.mark.skipif(not _CORPUS_SEEDS, reason="No corpus seeds present") |
| 755 | @pytest.mark.parametrize(("target_name", "seed_path"), _CORPUS_SEEDS) |
| 756 | def test_corpus_seed_does_not_crash(target_name: str, seed_path: str) -> None: |
| 757 | """Replay each seed file through its fuzz target without unhandled errors.""" |
| 758 | if atheris is None: |
| 759 | pytest.skip("Atheris not installed; skipping corpus replay") |
| 760 | target = next(t for t in FUZZ_TARGETS if t.__name__ == target_name) |
| 761 | data = __import__("pathlib").Path(seed_path).read_bytes() |
| 762 | target(data) |
| 763 | |
| 764 | |
| 765 | if __name__ == "__main__" and FUZZING: |
| 766 | atheris.instrument_all() |
| 767 | atheris.Setup(sys.argv, fuzz_dispatch) |
| 768 | atheris.Fuzz() |