|
41 | 41 | HTTPUnprocessableEntityException, |
42 | 42 | ) |
43 | 43 | from app.core.pagination import Pagination |
| 44 | +from app.core.security import crypto_timestamp_serializer |
44 | 45 | from app.sep.apps.framework.script_source import ScriptExecuteWrite |
| 46 | +from app.sep.artifact_constants import ARTIFACT_DOWNLOAD_SALT |
45 | 47 | from app.sep.snippets.config import snippets_settings, SnippetSudoOption |
46 | 48 | from app.sep.snippets.crud import SnippetManager |
47 | 49 | from app.sep.snippets.deps import build_snippet_execution_meta |
@@ -89,6 +91,34 @@ def _snippet_query( |
89 | 91 | ) |
90 | 92 |
|
91 | 93 |
|
| 94 | +def _normalize_snippet_source(url: str) -> tuple[str, dict[str, object]]: |
| 95 | + """Return the URL prefix plus the decoded token payload. |
| 96 | +
|
| 97 | + Keeps scheme, host, and path so parity tests still catch base-URL drift, |
| 98 | + while replacing the timed token with its payload so two mintings in |
| 99 | + different seconds still compare equal. |
| 100 | +
|
| 101 | + :param url: The signed artifact-download URL to normalize. |
| 102 | + :return: ``(prefix, payload)`` where ``prefix`` ends at the final slash |
| 103 | + and ``payload`` is the decoded token body. |
| 104 | + """ |
| 105 | + prefix, token = url.rsplit("/", 1) |
| 106 | + payload = crypto_timestamp_serializer.loads(token, salt=ARTIFACT_DOWNLOAD_SALT) |
| 107 | + return f"{prefix}/", payload |
| 108 | + |
| 109 | + |
| 110 | +def _meta_dump_decoded(meta: SnippetExecutionMeta) -> dict[str, object]: |
| 111 | + """Return ``meta.model_dump()`` with ``snippet_source`` timestamp-normalized. |
| 112 | +
|
| 113 | + :param meta: The execution meta whose signed URL should be normalized. |
| 114 | + :return: A dump identical to ``meta.model_dump()`` except ``snippet_source`` |
| 115 | + is ``(url_prefix, decoded_payload)``. |
| 116 | + """ |
| 117 | + d = meta.model_dump() |
| 118 | + d["snippet_source"] = _normalize_snippet_source(d["snippet_source"]) |
| 119 | + return d |
| 120 | + |
| 121 | + |
92 | 122 | def _framework_processed_body( |
93 | 123 | script: SnippetScript, body: ScriptExecuteWrite |
94 | 124 | ) -> ScriptExecuteWrite: |
@@ -480,7 +510,7 @@ async def test_meta_matches_legacy_path( |
480 | 510 | ) |
481 | 511 | legacy_meta = _legacy_execution_meta(script.snippet, body) |
482 | 512 |
|
483 | | - assert hook_meta.model_dump() == legacy_meta.model_dump() |
| 513 | + assert _meta_dump_decoded(hook_meta) == _meta_dump_decoded(legacy_meta) |
484 | 514 |
|
485 | 515 | async def test_extra_args_reach_command_via_new_schema_alias( |
486 | 516 | self, |
@@ -548,7 +578,7 @@ async def test_extra_args_still_bind_via_legacy_alias( |
548 | 578 | ) |
549 | 579 | legacy_meta = _legacy_execution_meta(script.snippet, legacy_body) |
550 | 580 |
|
551 | | - assert new_alias_meta.model_dump() == legacy_meta.model_dump() |
| 581 | + assert _meta_dump_decoded(new_alias_meta) == _meta_dump_decoded(legacy_meta) |
552 | 582 |
|
553 | 583 | @pytest.mark.parametrize("requested_sudo", [True, False]) |
554 | 584 | async def test_optional_sudo_toggle_is_honored( |
|
0 commit comments