diff --git a/README.md b/README.md index b61e9a9..660f002 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,15 @@ List tags for a repository: regshape tag list -i docker.io/library/alpine ``` +Retrieve every page of tags: + +```bash +regshape tag list -i docker.io/library/alpine --all +``` + +Use `--n` to set the all-pages request size and `--last` to start after a tag. +See the [tag command reference](specs/cli/tag.md) for pagination and output options. + ## Documentation ### Guides diff --git a/specs/cli/tag.md b/specs/cli/tag.md index b18a2a1..2d28582 100644 --- a/specs/cli/tag.md +++ b/specs/cli/tag.md @@ -13,10 +13,8 @@ that map to the OCI Distribution Spec endpoints: Tag creation and movement are implicit side-effects of `manifest put` and are not exposed here. -> **Note:** Until `libs/transport/` is implemented, all commands issue HTTP -> requests directly via the `requests` library using the existing -> `libs/auth/` helpers. Migration to `RegistryClient` is tracked as a -> follow-up task. +All requests use the shared `RegistryClient` transport and its authentication +and telemetry support. --- @@ -50,15 +48,16 @@ image reference syntax with the registry always embedded: ### `tag list` -List all tags for a repository, with optional pagination. +List one page of tags, or opt into automatic traversal of all remaining pages. #### Options | Option | Short | Type | Default | Description | |---|---|---|---|---| | `--image-ref` | `-i` | string | required | Repository reference with embedded registry (e.g., `registry/repo`, `registry/repo:tag`) | -| `--n` | | integer | none | Maximum number of tags to return per page | +| `--n` | | integer | none; 100 with `--all` | Tags requested per page; must be positive with `--all` | | `--last` | | string | none | Return tags lexicographically after this value (pagination cursor) | +| `--all` | | flag | false | Retrieve and aggregate every remaining page after the optional cursor | | `--json` | | flag | false | Output the raw `TagList` JSON object instead of one tag per line | | `--output` | `-o` | path | stdout | Write output to this file instead of stdout | | `--time-methods` | | flag | false | Print execution time for individual method calls | @@ -71,19 +70,52 @@ List all tags for a repository, with optional pagination. ignored if present). 2. Resolve credentials from the Docker credential store for the extracted registry. -3. Issue `GET /v2//tags/list`, appending `?n=` and/or - `?last=` when the respective options are provided. +3. Without `--all`, issue one `GET /v2//tags/list`, appending `?n=` + and/or `?last=` when provided. Existing single-page behavior, including + `--n 0`, is unchanged. + With `--all`, call `list_tags_all()` with a positive page size (default 100) + and the optional initial tag cursor. Follow next links when present; + otherwise continue full pages using the last returned tag value. 4. Parse the response body into a `TagList` model. 5. If `--json` is set, print the JSON object. Otherwise, print one tag per line (suitable for scripting with `xargs` or other tools). -6. If `--output` is given, write to that file instead of stdout. +6. If `--output` is given, write to that file instead of stdout. All-pages + results are emitted only after the entire traversal succeeds; a later-page + failure does not print a successful partial list or overwrite an output file. + Repeated tags, including partial overlap between pages, fail with exit code 1 + rather than producing duplicates or silently deduplicating the response. + A page that includes an excluded `--last` or server-provided cursor tag also + fails instead of returning that tag. + If concurrent repository changes cause overlap, retry once the listing is + stable. + +Pagination follows [OCI Distribution Specification v1.1.1, Listing +Tags](https://github.com/opencontainers/distribution-spec/blob/v1.1.1/spec.md#listing-tags). +`Link` is optional; when present, its `next` target takes precedence even for +short/empty pages. Without a next link, a full page advances using an exclusive +tag-value cursor. The effective page size can change in a server-provided link. +See the [library operation](../operations/tags.md#list_tags_all) for safety +checks and stopping rules. The 100-tag default is a client policy, not an OCI +requirement, and `--n` does not limit the total size of an all-pages result. +Registry and continuation URL authorities must use ASCII spelling; use the +punycode hostname for an international domain. + +As a client safety policy, `--all` fails on the **tenth consecutive empty page +with a next link**, even when continuation links change. Receiving tags resets +the counter; an empty page without a next link completes normally. This limit +is not an OCI requirement or a cap on total pages. Failure exits with code 1, +prints no partial list, and leaves output files untouched. +All-pages results and `--last`/linked tag cursors must also satisfy the OCI +tag-name grammar (at most 128 ASCII characters); malformed tags fail before +output or cursor encoding. See the library reference above for the exact grammar. #### Exit Codes | Code | Meaning | |---|---| | 0 | Success | -| 1 | HTTP error (repo not found, auth failure, connection error) | +| 1 | Registry/transport error, invalid response, or unsafe/non-progressing pagination | +| 2 | Usage error, including non-positive `--n` with `--all` | #### Examples @@ -91,9 +123,15 @@ List all tags for a repository, with optional pagination. # Log in first regshape auth login -r acr.io -u alice -# List all tags for a repository +# List one page using registry defaults regshape tag list -i acr.io/myrepo/myimage +# Retrieve every tag (100 requested per page by default) +regshape tag list -i acr.io/myrepo/myimage --all + +# Retrieve all remaining tags after v1.5, requesting 10 per page +regshape tag list -i acr.io/myrepo/myimage --all --n 10 --last v1.5 + # List with pagination (first 10 tags) regshape tag list -i acr.io/myrepo/myimage --n 10 @@ -105,6 +143,9 @@ regshape tag list -i acr.io/myrepo/myimage --json # Save tag list to a file regshape tag list -i acr.io/myrepo/myimage -o tags.txt + +# Save the complete aggregated response +regshape tag list -i acr.io/myrepo/myimage --all --json -o tags.json ``` #### Output Format @@ -134,6 +175,10 @@ v2.0 | Repository not found | `Error [acr.io/myrepo/myimage]: Registry error: repository not found` | | Auth failure | `Error [acr.io/myrepo/myimage]: Authentication failed for acr.io` | | Connection error | `Error [acr.io/myrepo/myimage]: ` | +| Invalid all-pages size | Usage error: `--n` must be positive with `--all` | +| Unsafe next link | `Error []: Unsafe or invalid tag pagination URL: ` | +| Repeated page/cursor | `Error []: Tag pagination repeated a page/cursor for /` | +| Empty-page limit | `Error []: Tag pagination returned 10 consecutive empty linked pages` | --- diff --git a/specs/operations/tags.md b/specs/operations/tags.md index 8515ac4..6dfef2b 100644 --- a/specs/operations/tags.md +++ b/specs/operations/tags.md @@ -14,6 +14,7 @@ It never calls `requests` directly — all HTTP traffic goes through a | Operation | Method | Endpoint | |---|---|---| | List tags | `GET` | `/v2/{repo}/tags/list[?n=&last=]` | +| List all remaining tags | `GET` | Same endpoint, following links or tag cursors | | Delete tag | `DELETE` | `/v2/{repo}/manifests/{tag}` | --- @@ -60,12 +61,129 @@ Fetches the tag list for a repository (single page). 3. Call `client.get(path, params=params)`. 4. Pass response to `_raise_for_list_error`. 5. Deserialize and return `TagList.from_json(response.text)`. - Re-raises `TagError` directly; wraps all other exceptions in `TagError`. + Re-raises `TagError` directly; wraps field validation errors in `TagError`. **Decorator:** `@track_time` --- +### `list_tags_all` + +```python +@track_scenario("tag list all") +def list_tags_all( + client: RegistryClient, + repo: str, + page_size: Optional[int] = None, + last: Optional[str] = None, +) -> TagList: +``` + +Returns one `TagList` containing all remaining pages, in server order and with +tag spelling unchanged. `page_size=None` uses 100 per request. An explicit size +must be a positive integer; invalid sizes fail before network access. `last` +is an optional exclusive starting tag value, not an offset. + +The protocol is defined by [OCI Distribution Specification v1.1.1, Listing +Tags](https://github.com/opencontainers/distribution-spec/blob/v1.1.1/spec.md#listing-tags). +The helper and its default size are RegShape conveniences, not OCI requirements. + +**Traversal:** + +1. Request the first page with `n` and, when provided, `last`. +2. Read pagination metadata from that page's returned response, not a separate + mutable client response slot. +3. Prefer its `Link` with relation `next`, including on a short or empty page. + Parse [RFC 5988](https://www.rfc-editor.org/rfc/rfc5988.html) link parameters + without splitting commas inside URI references or quoted values. Resolve + relative targets against the actual response URL and preserve linked query + values and path. Follow [RFC 3986 reference resolution](https://www.rfc-editor.org/rfc/rfc3986.html#section-5.2) + for dot segments and query inheritance without collapsing empty path + segments: `pages/token//part` and `pages/token/part` are distinct paths. +4. If there is no next link and the current request has a positive `n`, continue + a full-sized page using its last tag as the next exclusive cursor. Preserve + other query fields byte-for-byte, including opaque percent-encoded values + and bare flags; replace only `last`. Stop on a short/empty unlinked page; + exact multiples may need a final empty request. +5. Use the effective `n` of each request, not the initial size. A linked request + may change `n` or omit it: with no `n` and no next link, its response is the + remaining unbounded list. +6. Aggregate pages only on success. Propagate later-page failures without + returning a successful partial result. + +`Link` is optional in the OCI specification. Consequently, absence of a link +alone does not terminate a full-sized cursor-paginated page. Conversely, page +length alone cannot terminate a page with a next link. OCI permits `n=0` for +single-page listing (empty result, no link); `list_tags()` retains that behavior, +while all-pages mode rejects it as a client input policy. + +**Client safety policies:** + +- Continuations must remain on the configured scheme, hostname, and effective + port. Reject cross-origin links, scheme changes, embedded user credentials, + malformed ports, raw whitespace/control characters, and backslashes before + dispatch. Explicitly configured HTTP registries remain supported. + Authorities must be ASCII: use punycode for international domain names. + Reject Unicode authorities rather than allowing differences between + validator and HTTP-client domain-name mappings to bypass origin checks. +- Disable automatic redirects on pagination requests, including authentication + probes and retries in both transport modes. Validate each `Location` before + following a 301/302/303/307/308 redirect so a foreign authentication challenge + cannot receive stored credentials. Follow at most 30 redirects per page; + reject missing locations and redirect cycles. Ordinary transport requests + retain their existing automatic-redirect default; cached responses are + separated by redirect policy so an auto-follow result cannot bypass an + explicit opt-out. Track redirect hops separately from completed tag pages, + allowing a same-origin redirect to canonicalize query ordering without being + mistaken for a repeated page. +- If challenge discovery at `/v2/` redirects, both transport modes validate + and follow same-origin probe redirects explicitly before accepting a + challenge. Probe chains have the same cycle checks and 30-hop bound; unsafe + targets fail before any credential exchange. +- Sanitize URL-parser failures and suppress their raw exception context: + malformed URI authorities may contain credentials even when parsing fails + before the userinfo check. Requests may prepare `Response.next` even when + automatic redirects are disabled, so the shared request wrapper converts + pre-return parsing failures into sanitized transport errors before logging. +- Remove fragments for request/cycle identity. Detect repeated URLs/cursors and + nonempty pages containing no new tags; raise `TagError` rather than looping. + Empty pages with distinct valid continuation links may still be followed. + Normalize only known pagination parameters in page identity; opaque values + such as `%FF` and `%FE` must remain distinct. +- Fail with `TagError` on the **tenth consecutive empty page with a next link**, + before requesting its continuation, even if opaque links keep changing. + Receiving tags resets the counter; an empty page without a next link completes + normally, including after nine empty linked pages. Redirect responses neither + increment nor reset this counter. The limit is a RegShape safety policy, not + an OCI requirement or a cap on total pages; failure returns no partial result. +- Reject malformed or ambiguous pagination headers/parameters, conflicting next + links, and anchored next links whose alternate context is not supported. +- Reject invalid tag values and pages belonging to another repository. + Returned tags and initial/linked `last` cursors must match the + [OCI tag grammar](https://github.com/opencontainers/distribution-spec/blob/v1.1.1/spec.md#pulling-manifests): + `[a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}` (at most 128 ASCII characters). + Validate before rendering or cursor encoding, so malformed strings produce + `TagError`, not extra output lines or an unhandled encoding failure. + This does not reinterpret opaque continuation fields or change single-page + `list_tags()` behavior. +- Reject repeated tags within a page or across partially overlapping pages. + This is an all-pages consistency policy, not silent deduplication. +- Reject any returned tag equal to an initial or linked exclusive cursor, + even on a short page. Retain requested cursor boundaries across redirects. + +There is no snapshot guarantee if tags change while listing. Aggregation does +not re-sort, lower-case, or silently deduplicate returned tags. Concurrent +changes that cause overlapping pages fail explicitly; retry against a stable +repository rather than treating a duplicated list as a complete result. + +**Returns:** `TagList` + +**Errors:** `TagError` for invalid input, unsafe/non-progressing pagination, +malformed responses, or registry errors; `AuthError` for authentication failures; +transport exceptions propagate unchanged. + +--- + ### `delete_tag` ```python @@ -100,6 +218,16 @@ is routed through the manifests endpoint. ## Private Helpers +The single-page and all-pages operations share `_parse_tag_page()` for +response parsing/error handling. `_next_tag_link()`, `_tag_page_parameters()`, +`_tag_page_key()`, and `_safe_tag_page_url()` implement the tag-specific +continuation rules above. The shared +[`resolve_registry_url()`](../../src/regshape/libs/transport/urls.py) utility +preserves URI semantics and enforces the same origin policy for pagination +and authentication probes. +Requests are dispatched as registry-relative paths so both middleware and legacy +transport modes work. + ### `_raise_for_list_error` | HTTP status | Exception | Message | @@ -128,9 +256,9 @@ the first 200 characters of `response.text`. | Decorator | Applied to | |---|---| | `@track_time` | `list_tags`, `delete_tag` | +| `@track_scenario` | `list_tags_all` | -No `@track_scenario` is needed — neither operation is multi-step at the HTTP -level. +All page requests still use the shared transport's request tracing and metrics. --- @@ -144,6 +272,7 @@ level. | `400` / `405` on delete | `TagError` "not supported" | `_raise_for_delete_error` | | Malformed JSON body | `TagError` | `TagList.from_json` (re-raised) | | Transport / connection failure | `requests.exceptions.RequestException` | propagated as-is | +| Non-positive all-pages size, unsafe link, cycle, or invalid page | `TagError` | `list_tags_all` / pagination helpers | --- @@ -156,6 +285,7 @@ level. - `regshape.libs.refs` — `format_ref` - `regshape.libs.transport` — `RegistryClient` - `regshape.libs.decorators.timing` — `track_time` +- `regshape.libs.decorators.scenario` — `track_scenario` **External:** - `requests` — `requests.Response` type annotation only diff --git a/src/regshape/cli/tag.py b/src/regshape/cli/tag.py index 2b5f6c7..88b74ee 100644 --- a/src/regshape/cli/tag.py +++ b/src/regshape/cli/tag.py @@ -15,15 +15,14 @@ import click import requests -from regshape.cli.formatting import emit_error, emit_json, emit_list, emit_text +from regshape.cli.formatting import emit_error, emit_json, emit_list from regshape.libs.decorators import telemetry_options from regshape.libs.decorators.scenario import track_scenario from regshape.libs.errors import AuthError, TagError from regshape.libs.refs import format_ref, parse_image_ref -from regshape.libs.tags import delete_tag, list_tags +from regshape.libs.tags import delete_tag, list_tags, list_tags_all from regshape.libs.transport import RegistryClient, TransportConfig - # =========================================================================== # Public Click group # =========================================================================== @@ -57,7 +56,7 @@ def tag(): type=int, default=None, metavar="COUNT", - help="Maximum number of tags to return (pagination page size).", + help="Tags requested per page; with --all, must be positive (default: 100).", ) @click.option( "--last", @@ -68,6 +67,13 @@ def tag(): "Used together with --n to page through results." ), ) +@click.option( + "--all", + "fetch_all", + is_flag=True, + default=False, + help="Retrieve all remaining pages (default: one page only).", +) @click.option( "--json", "as_json", @@ -84,7 +90,15 @@ def tag(): ) @click.pass_context @track_scenario("tag list") -def tag_list(ctx, image_ref, page_size, last, as_json, output): +def tag_list( + ctx: click.Context, + image_ref: str, + page_size: int | None, + last: str | None, + fetch_all: bool, + as_json: bool, + output: str | None, +) -> None: """List tags for the repository identified by IMAGE_REF. IMAGE_REF must embed the registry (``registry/repo`` or @@ -92,12 +106,17 @@ def tag_list(ctx, image_ref, page_size, last, as_json, output): registry and repository namespace are used. Credentials are resolved automatically from the credential store populated by ``auth login``. - Use ``--n`` and ``--last`` to page through large tag lists. Use - ``--json`` to receive the full response object instead of one tag per - line. + Use ``--n`` and ``--last`` for one-page requests, or ``--all`` to retrieve + every remaining page after the optional cursor. All-pages mode uses 100 + tags per request by default and rejects non-positive page sizes. Use + ``--json`` for the full response object instead of one tag per line. + No results are emitted until the entire requested traversal succeeds. """ insecure = ctx.obj.get("insecure", False) if ctx.obj else False + if fetch_all and page_size is not None and page_size <= 0: + raise click.BadParameter("must be positive with --all", param_hint="--n") + try: registry, repo, _ = parse_image_ref(image_ref) except ValueError as exc: @@ -106,7 +125,8 @@ def tag_list(ctx, image_ref, page_size, last, as_json, output): client = RegistryClient(TransportConfig(registry=registry, insecure=insecure)) try: - tag_list = list_tags( + operation = list_tags_all if fetch_all else list_tags + result = operation( client=client, repo=repo, page_size=page_size, @@ -116,9 +136,9 @@ def tag_list(ctx, image_ref, page_size, last, as_json, output): emit_error(image_ref, str(exc)) if as_json: - emit_json(tag_list.to_dict(), output) + emit_json(result.to_dict(), output) else: - emit_list(tag_list.tags, output) + emit_list(result.tags, output) # =========================================================================== @@ -181,5 +201,3 @@ def delete(ctx, image_ref): # =========================================================================== # Internal helpers — output and error # =========================================================================== - - diff --git a/src/regshape/libs/decorators/call_details.py b/src/regshape/libs/decorators/call_details.py index cfa720a..c3ce07f 100644 --- a/src/regshape/libs/decorators/call_details.py +++ b/src/regshape/libs/decorators/call_details.py @@ -358,7 +358,7 @@ def wrapper(*args, **kwargs): @debug_call -def http_request(url: str, method: str = "GET", headers: dict = None, **kwargs): +def http_request(url: str, method: str = "GET", headers: dict = None, **kwargs) -> requests.Response: """Thin wrapper around :func:`requests.request` decorated with ``@debug_call``. @@ -371,5 +371,15 @@ def http_request(url: str, method: str = "GET", headers: dict = None, **kwargs): :param kwargs: Additional keyword arguments forwarded to :func:`requests.request` (e.g. ``timeout``). :return: :class:`requests.Response` + :raises requests.exceptions.RequestException: A sanitized transport failure + if Requests raises a parsing/argument ValueError with automatic redirects + disabled. This occurs before telemetry or middleware can log the error. """ - return requests.request(method, url, headers=headers or {}, **kwargs) + try: + return requests.request(method, url, headers=headers or {}, **kwargs) + except ValueError: + if kwargs.get("allow_redirects", True): + raise + raise requests.exceptions.RequestException( + "Invalid registry request or redirect response" + ) from None diff --git a/src/regshape/libs/models/tags.py b/src/regshape/libs/models/tags.py index e61e98c..02fd06c 100644 --- a/src/regshape/libs/models/tags.py +++ b/src/regshape/libs/models/tags.py @@ -33,9 +33,9 @@ class TagList: :param namespace: Repository namespace as returned in the ``"name"`` wire field (e.g. ``"myrepo/myimage"``). :param tags: Ordered list of tag strings. The OCI spec requires tags to be - in lexical ("ASCIIbetical") order; this class preserves the order as - received without re-sorting. Normalised from ``null`` to ``[]`` during - deserialization. + in lexical (case-insensitive alphanumeric) order; this class preserves + the order and spelling as received without re-sorting. Normalised + from ``null`` to ``[]`` during deserialization. """ namespace: str diff --git a/src/regshape/libs/tags/__init__.py b/src/regshape/libs/tags/__init__.py index 5b44f0c..1254a4c 100644 --- a/src/regshape/libs/tags/__init__.py +++ b/src/regshape/libs/tags/__init__.py @@ -14,9 +14,11 @@ from regshape.libs.tags.operations import ( delete_tag, list_tags, + list_tags_all, ) __all__ = [ "delete_tag", "list_tags", + "list_tags_all", ] diff --git a/src/regshape/libs/tags/operations.py b/src/regshape/libs/tags/operations.py index fa0cf27..e96a3b9 100644 --- a/src/regshape/libs/tags/operations.py +++ b/src/regshape/libs/tags/operations.py @@ -17,16 +17,38 @@ concerns — error reporting is the caller's responsibility. """ +import re from typing import Optional +from urllib.parse import ( + unquote_plus, + unquote_to_bytes, + urlencode, + urlsplit, + urlunsplit, +) import requests +from regshape.libs.decorators.scenario import track_scenario from regshape.libs.decorators.timing import track_time from regshape.libs.errors import AuthError, TagError from regshape.libs.models.error import OciErrorResponse from regshape.libs.models.tags import TagList from regshape.libs.refs import format_ref from regshape.libs.transport import RegistryClient +from regshape.libs.transport.urls import ( + MAX_REDIRECTS, + REDIRECT_STATUSES, + resolve_registry_url, +) + +_DEFAULT_PAGE_SIZE = 100 +_MAX_EMPTY_LINKED_PAGES = 10 +_OCI_TAG_RE = re.compile(r"[a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}") +_LINK_TARGET_RE = re.compile(r"\s*<([^<>]*)>") +_LINK_PARAMETER_RE = re.compile( + r'''\s*;\s*([!#$%&'*+.^_`|~0-9A-Za-z-]+)(?:\s*=\s*("(?:[^"\\]|\\.)*"|[^;,\s"]+))?''' +) # =========================================================================== @@ -65,18 +87,131 @@ def list_tags( params["last"] = last response = client.get(path, params=params if params else None) - _raise_for_list_error(response, client.config.registry, repo) + return _parse_tag_page(response, client.config.registry, repo) - try: - return TagList.from_json(response.text) - except TagError: - raise - except Exception as exc: - raise TagError( - f"Failed to parse tag-list response from " - f"{client.config.registry}/{repo}", - str(exc), - ) from exc + +@track_scenario("tag list all") +def list_tags_all( + client: RegistryClient, + repo: str, + page_size: Optional[int] = None, + last: Optional[str] = None, +) -> TagList: + """Fetch all remaining tags using OCI links or exclusive tag cursors. + + Prefer ``Link: rel="next"`` even on short or empty pages. Without a next + link, continue a full page using its last tag and the effective ``n`` of + that request. Preserve server ordering, spelling, and linked query values. + Follow at most 30 validated same-origin redirects per page, with automatic + redirects disabled before the transport processes authentication. + Fail on the tenth consecutive empty page with a next link; receiving tags + resets that counter. An empty page without a next link completes normally. + Validate returned tags and tag cursors against the OCI tag-name grammar. + No consistent snapshot is guaranteed if tags change during traversal. + + :param client: Transport client for the target registry. + :param repo: Repository namespace. + :param page_size: Positive initial page size; ``None`` uses 100. + :param last: Optional exclusive initial tag cursor. + :returns: One aggregated :class:`TagList`, only after traversal succeeds. + :raises TagError: For invalid input, malformed pages or links, unsafe + continuation origins, repeated pages/cursors, ten consecutive empty + linked pages, or registry errors. + :raises AuthError: On authentication failure on any page. + :raises requests.exceptions.RequestException: On transport failures. + """ + size = _DEFAULT_PAGE_SIZE if page_size is None else page_size + if isinstance(size, bool) or not isinstance(size, int) or size <= 0: + raise TagError("All-pages tag listing requires a positive page size") + if last is not None and (not isinstance(last, str) or not _OCI_TAG_RE.fullmatch(last)): + raise TagError("The initial tag cursor must be a valid OCI tag") + + params = [("n", str(size))] + if last is not None: + params.append(("last", last)) + url = _safe_tag_page_url( + client.base_url, + f"/v2/{repo}/tags/list?{urlencode(params)}", + client.base_url, + ) + tags: list[str] = [] + seen_tags: set[str] = set() + seen_pages: set[tuple[str, Optional[int], Optional[str], tuple[str, ...]]] = set() + used_cursors: set[str] = set() + redirect_hops: set[str] = set() + redirects = 0 + empty_linked_pages = 0 + + while True: + parts = urlsplit(url) + key = _tag_page_key(url) + requested_cursor = key[2] + if requested_cursor is not None: + used_cursors.add(requested_cursor) + if key in seen_pages: + raise TagError(f"Tag pagination repeated a page for {client.config.registry}/{repo}") + if url in redirect_hops: + raise TagError("Tag pagination redirect cycle detected") + redirect_hops.add(url) + + # Relative requests also work with the client's legacy transport mode. + path = urlunsplit(("", "", parts.path, parts.query, "")) + response = client.get(path, params=None, allow_redirects=False) + response_url = _safe_tag_page_url(client.base_url, response.url or url, url) + query, effective_size, cursor = _tag_page_parameters(response_url) + response_parts = urlsplit(response_url) + response_key = _tag_page_key(response_url) + if response_key != key and response_key in seen_pages: + raise TagError(f"Tag pagination revisited a page for {client.config.registry}/{repo}") + if response.status_code in REDIRECT_STATUSES: + if redirects >= MAX_REDIRECTS: + raise TagError("Tag pagination redirect limit exceeded") + location = response.headers.get("Location") + if not location: + raise TagError("Tag pagination redirect is missing Location") + url = _safe_tag_page_url(client.base_url, location, response_url) + redirects += 1 + continue + redirects = 0 + redirect_hops.clear() + page = _parse_tag_page(response, client.config.registry, repo) + seen_pages.update((key, response_key)) + if cursor is not None: + used_cursors.add(cursor) + if page.namespace != repo: + raise TagError("Tag pagination returned a different repository namespace") + if any(not isinstance(tag, str) or not _OCI_TAG_RE.fullmatch(tag) for tag in page.tags): + raise TagError("Tag pagination returned an invalid tag value") + page_tags = set(page.tags) + if page.tags and page_tags <= seen_tags: + raise TagError(f"Tag pagination made no progress for {client.config.registry}/{repo}") + if len(page_tags) != len(page.tags) or page_tags & seen_tags: + raise TagError(f"Tag pagination repeated a tag for {client.config.registry}/{repo}") + if page_tags & used_cursors: + raise TagError(f"Tag pagination repeated a cursor for {client.config.registry}/{repo}") + tags.extend(page.tags) + seen_tags.update(page_tags) + if page.tags: + empty_linked_pages = 0 + + next_link = _next_tag_link(response.headers.get("Link", "")) + if next_link is not None: + if not page.tags: + empty_linked_pages += 1 + if empty_linked_pages >= _MAX_EMPTY_LINKED_PAGES: + raise TagError( + f"Tag pagination returned {_MAX_EMPTY_LINKED_PAGES} " + "consecutive empty linked pages" + ) + url = _safe_tag_page_url(client.base_url, next_link, response_url) + continue + if effective_size is None or len(page.tags) < effective_size: + return TagList(namespace=repo, tags=tags) + + next_cursor = page.tags[-1] + fields = [field for name, field in query if name != b"last"] + fields.append(urlencode({"last": next_cursor})) + url = urlunsplit(response_parts._replace(query="&".join(fields))) @track_time @@ -108,6 +243,157 @@ def delete_tag( # =========================================================================== +def _parse_tag_page( + response: requests.Response, + registry: str, + repo: str, +) -> TagList: + """Parse one page using the shared tag error mapping. + + The caller retains the response for pagination and validates any redirects + before parsing. Parsing failures use :class:`TagError`; authentication + failures retain their :class:`AuthError` mapping. + """ + _raise_for_list_error(response, registry, repo) + try: + page = TagList.from_json(response.text) + except TagError: + raise + except (TypeError, ValueError) as exc: + raise TagError( + f"Failed to parse tag-list response from {registry}/{repo}", + str(exc), + ) from exc + return page + + +def _safe_tag_page_url(base_url: str, target: str, context_url: str) -> str: + """Resolve a continuation against its response URL and enforce same origin. + + Reject userinfo, cross-origin links, scheme changes, and ambiguous control + characters before dispatch. Fragments are not sent in HTTP requests and + are removed so they cannot conceal pagination cycles. Invalid URLs raise + :class:`TagError` without exposing credential-bearing link values, parser + diagnostics, or the original exception chain. + """ + try: + return resolve_registry_url(base_url, target, context_url) + except ValueError: + raise TagError( + "Unsafe or invalid tag pagination URL", + "URLs must use a valid ASCII authority, contain no credentials, " + "and stay on the configured registry origin", + ) from None + + +def _tag_page_parameters( + url: str, +) -> tuple[list[tuple[bytes, str]], Optional[int], Optional[str]]: + """Return raw query fields, the effective page size, and the tag cursor. + + Decode field names as bytes only to identify ``n`` and ``last``; leave + opaque values and their encoding untouched. Decode pagination values + strictly, rejecting invalid encodings, duplicate parameters, and + non-positive sizes rather than corrupting or truncating a traversal. + Cursors must match the OCI tag-name grammar; opaque fields are not tags. + """ + raw_query = urlsplit(url).query + fields = raw_query.split("&") if raw_query else [] + query = [ + (unquote_to_bytes(field.partition("=")[0].replace("+", " ")), field) + for field in fields + ] + try: + sizes = [ + unquote_plus(field.partition("=")[2], errors="strict") + for name, field in query if name == b"n" + ] + cursors = [ + unquote_plus(field.partition("=")[2], errors="strict") + for name, field in query if name == b"last" + ] + except UnicodeError as exc: + raise TagError("Invalid tag pagination parameter encoding") from exc + if len(sizes) > 1 or len(cursors) > 1: + raise TagError("Ambiguous tag pagination parameters") + size = None + if sizes: + if not sizes[0].isascii() or not sizes[0].isdecimal(): + raise TagError("Tag pagination page size must be a positive integer") + try: + size = int(sizes[0]) + except ValueError as exc: + raise TagError("Tag pagination page size must be a positive integer") from exc + if size <= 0: + raise TagError("Tag pagination page size must be a positive integer") + if cursors and not _OCI_TAG_RE.fullmatch(cursors[0]): + raise TagError("Tag pagination cursor must be a valid OCI tag") + return query, size, cursors[0] if cursors else None + + +def _tag_page_key( + url: str, +) -> tuple[str, Optional[int], Optional[str], tuple[str, ...]]: + """Identify a page without conflating distinct opaque continuation bytes. + + Normalize only the understood pagination parameters. Opaque fields retain + their original order and encoding, including repeated keys and bare flags. + Invalid pagination parameters propagate as :class:`TagError`. + """ + query, size, cursor = _tag_page_parameters(url) + opaque_fields = tuple(field for name, field in query if name not in (b"n", b"last")) + return urlsplit(url).path, size, cursor, opaque_fields + + +def _next_tag_link(header: str) -> Optional[str]: + """Select an RFC 5988 next link without splitting quoted values or URI commas. + + Unknown extension parameters are ignored. Honor only the first ``rel`` + parameter of each link, as required by the RFC. Malformed headers, + conflicting next links, and unsupported anchored next links raise + :class:`TagError` rather than risking incomplete or misdirected traversal. + """ + position = 0 + next_url = None + while position < len(header): + while position < len(header) and header[position] in " \t,": + position += 1 + if position == len(header): + break + target = _LINK_TARGET_RE.match(header, position) + if target is None: + raise TagError("Malformed tag pagination Link header") + position = target.end() + relations = None + anchored = False + while True: + parameter = _LINK_PARAMETER_RE.match(header, position) + if parameter is None: + break + name, value = parameter.groups() + name = name.lower() + if value is not None and value.startswith('"'): + value = re.sub(r"\\(.)", r"\1", value[1:-1]) + if name == "rel" and relations is None: + if value is None or not value.strip(): + raise TagError("Tag pagination Link relation is missing its value") + relations = value + anchored = anchored or name == "anchor" + position = parameter.end() + while position < len(header) and header[position].isspace(): + position += 1 + if position < len(header) and header[position] != ",": + raise TagError("Malformed tag pagination Link parameters") + if relations is not None and "next" in relations.lower().split(): + if anchored: + raise TagError("Anchored tag pagination links are not supported") + if next_url is not None and next_url != target.group(1): + raise TagError("Ambiguous next links in tag pagination response") + next_url = target.group(1) + position += 1 + return next_url + + def _raise_for_list_error( response: requests.Response, registry: str, diff --git a/src/regshape/libs/transport/client.py b/src/regshape/libs/transport/client.py index a619460..5ef6057 100644 --- a/src/regshape/libs/transport/client.py +++ b/src/regshape/libs/transport/client.py @@ -169,6 +169,7 @@ def _setup_middleware_pipeline(self) -> MiddlewarePipeline: username=self._username, password=self._password, registry=self.config.registry, + base_url=self.base_url, ) ) @@ -197,12 +198,14 @@ def _terminal_handler(self, request: RegistryRequest) -> RegistryResponse: """Terminal handler that performs the actual HTTP request. This is the final handler in the middleware pipeline that converts - RegistryRequest to a requests call and wraps the response. + RegistryRequest to a requests call and wraps the response. Preserve + an explicit redirect opt-out before authentication sees the response. """ # Convert RegistryRequest to requests parameters url = f"{self.base_url}{request.url}" if request.url.startswith('/') else request.url # Use http_request for telemetry (--debug-calls) + redirect_options = {} if request.allow_redirects else {"allow_redirects": False} response = http_request( url=url, method=request.method, @@ -210,7 +213,8 @@ def _terminal_handler(self, request: RegistryRequest) -> RegistryResponse: data=request.body, stream=request.stream, params=request.params, - timeout=request.timeout or self.config.timeout + timeout=request.timeout or self.config.timeout, + **redirect_options, ) # Store for backward compatibility @@ -230,7 +234,7 @@ def _legacy_authenticate_and_retry( timeout: int, **kwargs ) -> requests.Response: - """Legacy authentication handling for when middleware is disabled.""" + """Handle legacy authentication, retaining redirect policy on probes/retries.""" response = http_request(url, method, headers=req_headers, timeout=timeout, **kwargs) self.last_response = response @@ -242,10 +246,19 @@ def _legacy_authenticate_and_retry( if not www_auth: # Some registries (e.g. ACR) only return WWW-Authenticate on # the /v2/ endpoint. Fall back to a /v2/ probe. - v2_url = f"{self.base_url}/v2/" - v2_resp = http_request(v2_url, "GET", headers={}, timeout=timeout) - if v2_resp.status_code == 401: - www_auth = v2_resp.headers.get("WWW-Authenticate", "") + if kwargs.get("allow_redirects", True): + v2_url = f"{self.base_url}/v2/" + v2_resp = http_request(v2_url, "GET", headers={}, timeout=timeout) + if v2_resp.status_code == 401: + www_auth = v2_resp.headers.get("WWW-Authenticate", "") + else: + original_request = RegistryRequest( + method=method, url=url, headers=req_headers, + timeout=timeout, allow_redirects=False, + ) + www_auth = AuthMiddleware._probe_v2_challenge( + self._terminal_handler, original_request, self.base_url, + ) if not www_auth: raise AuthError( "Authentication failed", @@ -303,7 +316,9 @@ def request( :param headers: Optional request headers. A shallow copy is taken so that the caller dict is never mutated. :param kwargs: Additional keyword arguments forwarded to the HTTP - request (e.g. data, params, stream). + request (e.g. data, params, stream). ``allow_redirects=False`` + disables automatic redirects, including authentication probes + and retries, in both transport modes. :returns: The HTTP response from the registry. :raises AuthError: If authentication fails. :raises requests.exceptions.RequestException: On transport errors. @@ -320,7 +335,8 @@ def request( body=kwargs.get('data'), stream=kwargs.get('stream', False), params=kwargs.get('params'), - timeout=timeout + timeout=timeout, + allow_redirects=kwargs.get("allow_redirects", True), ) try: diff --git a/src/regshape/libs/transport/middleware.py b/src/regshape/libs/transport/middleware.py index 344098d..77df4d5 100644 --- a/src/regshape/libs/transport/middleware.py +++ b/src/regshape/libs/transport/middleware.py @@ -14,11 +14,13 @@ """ from abc import ABC +from collections.abc import Mapping from typing import Protocol, Callable from regshape.libs.auth import registryauth from regshape.libs.errors import AuthError from regshape.libs.transport.models import RegistryRequest, RegistryResponse +from regshape.libs.transport.urls import MAX_REDIRECTS, REDIRECT_STATUSES, resolve_registry_url class Middleware(Protocol): @@ -255,10 +257,10 @@ def wrapped_handler(request: RegistryRequest) -> RegistryResponse: # Internal helpers # --------------------------------------------------------------------------- -def _get_header_ci(headers: dict, name: str, default: str = "") -> str: +def _get_header_ci(headers: Mapping[str, str], name: str, default: str = "") -> str: """Case-insensitive header lookup. - ``RegistryResponse.headers`` is a plain ``dict`` whose keys preserve + ``RegistryResponse.headers`` is a mapping whose keys preserve the original casing from the server (e.g. ``Www-Authenticate`` from Azure Container Registry). This helper performs a case-insensitive search so that header checks work regardless of server casing. @@ -348,6 +350,8 @@ class AuthMiddleware(BaseMiddleware): :param username: Username for authentication, or ``None``. :param password: Password for authentication, or ``None``. :param registry: Registry hostname, used only in error messages. + :param base_url: Configured registry origin for validating probe redirects. + When omitted, an absolute original request URL can supply the origin. """ def __init__( @@ -355,10 +359,13 @@ def __init__( username: Optional[str] = None, password: Optional[str] = None, registry: str = "", - ): + base_url: Optional[str] = None, + ) -> None: + """Configure credentials and the origin used for challenge discovery.""" self._username = username self._password = password self._registry = registry + self._base_url = base_url # -- Override __call__ to get access to next_handler for the retry ------ @@ -367,7 +374,7 @@ def __call__( request: RegistryRequest, next_handler: Callable[[RegistryRequest], RegistryResponse], ) -> RegistryResponse: - """Execute the request and handle 401 challenges.""" + """Handle 401 challenges while retaining the request's redirect policy.""" processed_request = request try: processed_request = self.process_request(request) @@ -384,7 +391,9 @@ def __call__( # Some registries (e.g. ACR) only return WWW-Authenticate on # the /v2/ endpoint. Fall back to a /v2/ probe to obtain the # challenge, then retry the original request. - www_auth = self._probe_v2_challenge(next_handler, processed_request) + www_auth = self._probe_v2_challenge( + next_handler, processed_request, self._base_url, + ) if not www_auth: raise AuthError( "Authentication failed", @@ -421,6 +430,7 @@ def __call__( stream=processed_request.stream, params=processed_request.params, timeout=processed_request.timeout, + allow_redirects=processed_request.allow_redirects, ) return next_handler(retry_request) @@ -431,27 +441,71 @@ def __call__( def _probe_v2_challenge( next_handler: Callable[[RegistryRequest], RegistryResponse], original_request: RegistryRequest, + base_url: Optional[str] = None, ) -> str: """Issue ``GET /v2/`` to obtain a ``WWW-Authenticate`` challenge. Some registries only return the challenge header on the ``/v2/`` base endpoint rather than on every 401 response. This helper sends a lightweight probe so that the caller can still negotiate - authentication. + authentication. The probe inherits the original redirect policy so + an unvalidated redirect cannot introduce a foreign challenge. When + automatic redirects are disabled, follow only validated same-origin + targets, with cycle detection and a 30-hop limit. :returns: The ``WWW-Authenticate`` header value, or an empty string if the probe also lacks one. + :raises AuthError: For unsafe, malformed, or non-terminating probe redirects. """ probe = RegistryRequest( method="GET", url="/v2/", headers={}, timeout=original_request.timeout, + allow_redirects=original_request.allow_redirects, ) - probe_response = next_handler(probe) - if probe_response.status_code == 401: - return _get_header_ci(probe_response.headers, "WWW-Authenticate") - return "" + if original_request.allow_redirects: + probe_response = next_handler(probe) + if probe_response.status_code == 401: + return _get_header_ci(probe_response.headers, "WWW-Authenticate") + return "" + + origin = base_url or original_request.url + try: + current_url = resolve_registry_url(origin, "/v2/", origin) + except ValueError: + raise AuthError("Cannot safely resolve the authentication probe origin") from None + visited: set[str] = set() + redirects = 0 + while True: + if current_url in visited: + raise AuthError("Authentication probe redirect cycle detected") + visited.add(current_url) + probe_response = next_handler(probe) + try: + response_url = resolve_registry_url( + origin, probe_response.raw_response.url or current_url, current_url, + ) + except ValueError: + raise AuthError("Unsafe authentication probe response origin") from None + if probe_response.status_code not in REDIRECT_STATUSES: + if probe_response.status_code == 401: + return _get_header_ci(probe_response.headers, "WWW-Authenticate") + return "" + if redirects >= MAX_REDIRECTS: + raise AuthError("Authentication probe redirect limit exceeded") + location = _get_header_ci(probe_response.headers, "Location") + if not location: + raise AuthError("Authentication probe redirect is missing Location") + try: + current_url = resolve_registry_url(origin, location, response_url) + except ValueError: + raise AuthError("Unsafe authentication probe redirect") from None + probe = RegistryRequest( + method="GET", url=current_url, headers={}, + timeout=original_request.timeout, allow_redirects=False, + ) + redirects += 1 class LoggingMiddleware(BaseMiddleware): @@ -580,7 +634,8 @@ def _wait_backoff(self, attempt: int) -> None: class CachingMiddleware(BaseMiddleware): """Middleware that caches GET responses for immutable content. - Caches responses based on URL, Accept header, and query parameters. + Caches responses based on URL, Accept header, query parameters, and + redirect policy. Suitable for registry manifests and blobs that are content-addressed and therefore immutable. @@ -636,8 +691,8 @@ def _get_cache_key(self, request: RegistryRequest) -> str: """Generate cache key from request. Includes HTTP method, URL, relevant headers (e.g. Accept), - and query parameters (if available) to avoid collisions - between requests that can yield different representations. + query parameters, and redirect policy to avoid collisions between + requests that can yield different representations. """ # Base components method = getattr(request, "method", "") @@ -658,7 +713,8 @@ def _get_cache_key(self, request: RegistryRequest) -> str: sorted_items = sorted(params.items()) params_component = "&".join(f"{k}={v}" for k, v in sorted_items) - return f"{method}:{url}:accept={accept}:params={params_component}" + redirects = getattr(request, "allow_redirects", True) + return f"{method}:{url}:accept={accept}:params={params_component}:redirects={redirects}" def _should_cache(self, response: RegistryResponse) -> bool: """Determine if response should be cached.""" diff --git a/src/regshape/libs/transport/models.py b/src/regshape/libs/transport/models.py index 337c7a2..fbe2a81 100644 --- a/src/regshape/libs/transport/models.py +++ b/src/regshape/libs/transport/models.py @@ -35,6 +35,8 @@ class RegistryRequest: :param stream: Whether to stream the response body :param params: Query parameters as key-value dict :param timeout: Request timeout in seconds + :param allow_redirects: Whether the HTTP layer may follow redirects. + Defaults to True; False is preserved through authentication retries. """ method: str url: str @@ -43,6 +45,7 @@ class RegistryRequest: stream: bool = False params: Optional[Dict[str, Any]] = None timeout: Optional[int] = None + allow_redirects: bool = True def __post_init__(self) -> None: """Validate request fields.""" diff --git a/src/regshape/libs/transport/urls.py b/src/regshape/libs/transport/urls.py new file mode 100644 index 0000000..63b0f14 --- /dev/null +++ b/src/regshape/libs/transport/urls.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 + +"""Shared URI resolution and origin checks for registry pagination and auth probes. + +Redirect users share the same 30-hop safety budget and supported status codes. +These client policies do not change the default behavior of ordinary requests. +""" + +from urllib.parse import urlsplit, urlunsplit + +MAX_REDIRECTS = 30 +REDIRECT_STATUSES = (301, 302, 303, 307, 308) + + +def resolve_registry_url(base_url: str, target: str, context_url: str) -> str: + """Resolve a credential-free HTTP reference on the configured registry origin. + + Preserve URI path/query semantics, remove fragments, and reject scheme or + origin changes before dispatch. Authorities must be ASCII; international + domain names must use their punycode form to avoid differing IDNA mappings + between URL validation and the HTTP client. Invalid references raise a + sanitized error without retaining a credential-bearing exception chain. + + :param base_url: Configured registry origin. + :param target: Absolute or relative URI reference from a response. + :param context_url: Actual response/request URL used as the reference base. + :returns: An absolute URL with the same scheme, host, and effective port. + :raises ValueError: For invalid, credential-bearing, or cross-origin URLs. + """ + try: + if any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in target): + raise ValueError("Invalid URI characters") + if "\\" in target: + raise ValueError("Invalid URI backslash") + resolved = _resolve_reference(context_url, target) + if _url_origin(resolved) != _url_origin(base_url): + raise ValueError("Registry origin changed") + parts = urlsplit(resolved) + return urlunsplit(parts._replace(path=parts.path or "/")) + except ValueError: + raise ValueError( + "URLs must use a valid ASCII authority, contain no credentials, " + "and stay on the configured registry origin" + ) from None + + +def _url_origin(url: str) -> tuple[str, str, int]: + """Return an ASCII HTTP origin; reject Unicode authorities, userinfo, and bad ports.""" + parts = urlsplit(url) + if parts.scheme not in ("http", "https") or not parts.hostname: + raise ValueError("HTTP origin required") + if not parts.netloc.isascii(): + raise ValueError("ASCII URI authority required") + if parts.username is not None or parts.password is not None: + raise ValueError("URI credentials are not allowed") + port = parts.port + if port is None: + port = 443 if parts.scheme == "https" else 80 + return parts.scheme, parts.hostname, port + + +def _resolve_reference(context_url: str, target: str) -> str: + """Resolve RFC 3986 references without collapsing empty path segments. + + Distinguish an absent query from an explicitly empty query. Fragments are + removed because they are not part of an HTTP request target. + """ + base = urlsplit(context_url) + reference = urlsplit(target) + if reference.scheme: + resolved = reference._replace(path=_remove_dot_segments(reference.path)) + elif reference.netloc or target.startswith("//"): + resolved = reference._replace( + scheme=base.scheme, + path=_remove_dot_segments(reference.path), + ) + else: + if not reference.path: + path = base.path + query = reference.query if "?" in target.split("#", 1)[0] else base.query + else: + path = reference.path + if not path.startswith("/"): + path = base.path.rpartition("/")[0] + "/" + path + path = _remove_dot_segments(path) + query = reference.query + resolved = reference._replace( + scheme=base.scheme, netloc=base.netloc, path=path, query=query, + ) + return urlunsplit(resolved._replace(fragment="")) + + +def _remove_dot_segments(path: str) -> str: + """Apply RFC 3986 section 5.2.4 while retaining ordinary empty path segments.""" + output = "" + while path: + if path.startswith("../"): + path = path[3:] + elif path.startswith("./"): + path = path[2:] + elif path.startswith("/./"): + path = path[2:] + elif path == "/.": + path = "/" + elif path.startswith("/../"): + path = path[3:] + output = output.rpartition("/")[0] + elif path == "/..": + path = "/" + output = output.rpartition("/")[0] + elif path in (".", ".."): + path = "" + else: + boundary = path.find("/", 1 if path.startswith("/") else 0) + if boundary == -1: + output += path + path = "" + else: + output += path[:boundary] + path = path[boundary:] + return output diff --git a/src/regshape/tests/test_tag_pagination.py b/src/regshape/tests/test_tag_pagination.py new file mode 100644 index 0000000..a3f3659 --- /dev/null +++ b/src/regshape/tests/test_tag_pagination.py @@ -0,0 +1,1088 @@ +#!/usr/bin/env python3 + +"""OCI tag-pagination regressions across library, CLI, and transport boundaries.""" + +import json +import logging +import traceback +from collections.abc import Iterable +from pathlib import Path +from unittest.mock import MagicMock, call, patch + +import pytest +import requests +from click.testing import CliRunner + +from regshape.cli.main import regshape +from regshape.libs.decorators import configure_telemetry, get_telemetry_config +from regshape.libs.decorators.call_details import http_request +from regshape.libs.errors import AuthError, TagError +from regshape.libs.models.tags import TagList +from regshape.libs.tags import list_tags, list_tags_all +from regshape.libs.tags.operations import _next_tag_link, _safe_tag_page_url +from regshape.libs.transport import RegistryClient, TransportConfig + +REGISTRY = "registry.example.com" +REPO = "team/image" +PATH = f"/v2/{REPO}/tags/list" +BASE = f"https://{REGISTRY}" + + +def _page( + tags: object, + *, + link: str | None = None, + status: int = 200, + url: str | None = None, + namespace: str = REPO, +) -> requests.Response: + """Build a real in-memory response, including intentionally invalid tag data.""" + response = requests.Response() + response.status_code = status + response._content = json.dumps({"name": namespace, "tags": tags}).encode() + response.headers["Content-Type"] = "application/json" + if link is not None: + response.headers["Link"] = link + if url is not None: + response.url = url + return response + + +def _client( + pages: Iterable[requests.Response | Exception], *, insecure: bool = False +) -> MagicMock: + """Provide deterministic pages without credential lookup or network access.""" + client = MagicMock(spec=RegistryClient) + client.config = TransportConfig(registry=REGISTRY, insecure=insecure) + client.base_url = f"{'http' if insecure else 'https'}://{REGISTRY}" + client.get.side_effect = pages + return client + + +def _empty_linked_pages(count: int, *, start: int = 0) -> list[requests.Response]: + """Build empty pages with alternating cursors and distinct opaque next links.""" + return [ + _page([], link=f'; rel="next"') + for index in range(start, start + count) + ] + + +def _redirect(location: str | None, status: int = 302) -> requests.Response: + """Build a redirect without contacting or authenticating to its target.""" + response = _page([], status=status) + if location is not None: + response.headers["Location"] = location + return response + + +def _malformed_redirect_adapter( + request: requests.PreparedRequest, **_kwargs: object +) -> requests.Response: + """Return a wire-style redirect while exercising real Requests parsing.""" + location = f"https://user:synthetic-secret@{REGISTRY}\uff0fignored/tags" + response = _redirect(location.encode("utf-8").decode("latin-1")) + assert request.url is not None + response.url = request.url + response.request = request + return response + + +class TestTagRedirectSafety: + """Ensure redirected challenges and malformed URLs cannot expose credentials.""" + + @pytest.mark.parametrize("middleware", [False, True]) + def test_off_origin_redirect_cannot_trigger_credential_exchange(self, middleware: bool) -> None: + """Model Requests following a redirect unless automatic redirects are disabled.""" + redirected_challenge = _page([], status=401, url="https://other.example/tags") + redirected_challenge.headers["WWW-Authenticate"] = ( + 'Bearer realm="https://other.example/token",service="other"' + ) + + def respond(method: str, url: str, **kwargs: object) -> requests.Response: + """Expose the unsafe challenge only when the transport follows redirects.""" + if kwargs.get("allow_redirects", True): + return redirected_challenge + return _redirect("https://other.example/tags") + + with ( + patch("regshape.libs.decorators.call_details.requests.request", side_effect=respond) as http, + patch("regshape.libs.auth.registryauth.authenticate", return_value="test-token") as exchange, + ): + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_middleware=middleware, + )) + with pytest.raises(TagError, match="pagination URL"): + list_tags_all(client, REPO) + exchange.assert_not_called() + assert http.call_count == 1 + assert http.call_args.kwargs["allow_redirects"] is False + + @pytest.mark.parametrize("middleware", [False, True]) + @pytest.mark.parametrize("probe", [False, True]) + def test_authentication_probe_and_retry_keep_redirects_disabled( + self, middleware: bool, probe: bool + ) -> None: + """Preserve the request policy through implicit probes and authenticated retries.""" + challenge = _page([], status=401) + challenge.headers["WWW-Authenticate"] = 'Basic realm="registry"' + pages = [_page([], status=401)] if probe else [] + pages.extend([challenge, _redirect("https://other.example/tags")]) + with ( + patch("regshape.libs.decorators.call_details.requests.request", side_effect=pages) as http, + patch("regshape.libs.auth.registryauth.authenticate", return_value="test-token") as exchange, + ): + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_middleware=middleware, + )) + with pytest.raises(TagError): + list_tags_all(client, REPO) + exchange.assert_called_once() + assert all(entry.kwargs.get("allow_redirects", True) is False for entry in http.call_args_list) + + @pytest.mark.parametrize("middleware", [False, True]) + def test_redirected_auth_probe_cannot_trigger_exchange(self, middleware: bool) -> None: + """Do not allow the fallback /v2/ probe to bypass the redirect policy.""" + foreign_challenge = _page([], status=401, url="https://other.example/v2/") + foreign_challenge.headers["WWW-Authenticate"] = ( + 'Bearer realm="https://other.example/token",service="other"' + ) + + def respond(method: str, url: str, **kwargs: object) -> requests.Response: + """Return the probe redirect, or its challenge when auto-follow is enabled.""" + if url.endswith("/v2/"): + if kwargs.get("allow_redirects", True): + return foreign_challenge + return _redirect("https://other.example/v2/") + return _page([], status=401) + + with ( + patch("regshape.libs.decorators.call_details.requests.request", side_effect=respond) as http, + patch("regshape.libs.auth.registryauth.authenticate", return_value="test-token") as exchange, + ): + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_middleware=middleware, + )) + with pytest.raises(AuthError): + list_tags_all(client, REPO) + exchange.assert_not_called() + assert http.call_count == 2 + assert all(entry.kwargs.get("allow_redirects", True) is False for entry in http.call_args_list) + + @pytest.mark.parametrize("middleware", [False, True]) + @pytest.mark.parametrize("insecure", [False, True]) + def test_auth_probe_follows_validated_same_origin_redirect( + self, middleware: bool, insecure: bool + ) -> None: + """Find a challenge behind a safe redirect without re-enabling auto-follow.""" + challenge = _page([], status=401) + challenge.headers["WWW-Authenticate"] = 'Basic realm="registry"' + pages = [_page([], status=401), _redirect("challenge"), challenge, _page(["a"])] + with ( + patch("regshape.libs.decorators.call_details.requests.request", side_effect=pages) as http, + patch("regshape.libs.auth.registryauth.authenticate", return_value="test-token") as exchange, + ): + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_middleware=middleware, insecure=insecure, + )) + assert list_tags_all(client, REPO, page_size=2).tags == ["a"] + scheme = "http" if insecure else "https" + assert [entry.args[1] for entry in http.call_args_list] == [ + f"{scheme}://{REGISTRY}{PATH}?n=2", + f"{scheme}://{REGISTRY}/v2/", + f"{scheme}://{REGISTRY}/v2/challenge", + f"{scheme}://{REGISTRY}{PATH}?n=2", + ] + assert all(entry.kwargs.get("allow_redirects", True) is False for entry in http.call_args_list) + exchange.assert_called_once() + + @pytest.mark.parametrize("middleware", [False, True]) + @pytest.mark.parametrize("target", [ + "https://other.example/challenge", + f"http://{REGISTRY}/challenge", + f"https://user:synthetic-secret@{REGISTRY}\uff0fignored", + ]) + def test_auth_probe_validates_every_redirect(self, middleware: bool, target: str) -> None: + """A safe first hop must not authorize an unsafe subsequent challenge.""" + pages = [_page([], status=401), _redirect("challenge"), _redirect(target)] + with ( + patch("regshape.libs.decorators.call_details.requests.request", side_effect=pages) as http, + patch("regshape.libs.auth.registryauth.authenticate") as exchange, + ): + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_middleware=middleware, + )) + with pytest.raises(AuthError) as error: + list_tags_all(client, REPO) + assert "synthetic-secret" not in str(error.value) + assert http.call_count == 3 + exchange.assert_not_called() + + @pytest.mark.parametrize("middleware", [False, True]) + def test_auth_probe_rejects_unicode_host_before_dispatch(self, middleware: bool) -> None: + """Do not let incompatible IDNA mappings authorize a request to another host.""" + foreign = _page([], status=401, url="https://xn--fa-hia.example/challenge") + foreign.headers["WWW-Authenticate"] = 'Basic realm="other"' + pages = [ + _page([], status=401), + _redirect("https://fa\u00df.example/challenge"), + foreign, + ] + with ( + patch("regshape.libs.decorators.call_details.requests.request", side_effect=pages) as http, + patch("regshape.libs.auth.registryauth.authenticate") as exchange, + ): + client = RegistryClient(TransportConfig( + registry="fass.example", username="user", password="test-password", + enable_middleware=middleware, + )) + with pytest.raises(AuthError): + list_tags_all(client, REPO) + assert http.call_count == 2 + exchange.assert_not_called() + + @pytest.mark.parametrize("middleware", [False, True]) + @pytest.mark.parametrize("cycle", [False, True]) + def test_auth_probe_redirects_are_bounded(self, middleware: bool, cycle: bool) -> None: + """Bound both repeated probe URLs and distinct redirect chains before exchange.""" + redirects = ( + [_redirect("challenge"), _redirect("/v2/")] + if cycle else [_redirect(f"?hop={index}") for index in range(1, 33)] + ) + with ( + patch("regshape.libs.decorators.call_details.requests.request", + side_effect=[_page([], status=401), *redirects]) as http, + patch("regshape.libs.auth.registryauth.authenticate") as exchange, + ): + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_middleware=middleware, + )) + with pytest.raises(AuthError, match="redirect"): + list_tags_all(client, REPO) + assert http.call_count == (3 if cycle else 32) + exchange.assert_not_called() + + @pytest.mark.parametrize("middleware", [False, True]) + def test_adapter_redirect_parse_errors_are_sanitized_before_logging( + self, middleware: bool, caplog: pytest.LogCaptureFixture + ) -> None: + """Exercise Requests Response.next parsing rather than mocking requests.request.""" + with ( + patch("requests.adapters.HTTPAdapter.send", side_effect=_malformed_redirect_adapter) as send, + patch("regshape.libs.auth.registryauth.authenticate") as exchange, + caplog.at_level(logging.ERROR, logger="regshape.transport"), + ): + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_middleware=middleware, enable_logging=True, + )) + with pytest.raises(requests.exceptions.RequestException) as error: + list_tags_all(client, REPO) + rendered = "".join(traceback.format_exception( + type(error.value), error.value, error.value.__traceback__, + )) + assert "synthetic-secret" not in str(error.value) + assert "synthetic-secret" not in rendered + assert "synthetic-secret" not in caplog.text + assert error.value.__suppress_context__ + assert send.call_count == 1 + exchange.assert_not_called() + + def test_cli_handles_adapter_redirect_parse_errors(self, tmp_path: Path) -> None: + """Emit a safe CLI error and telemetry log for pre-return parser failures.""" + log_file = tmp_path / "trace.log" + previous = get_telemetry_config() + try: + with ( + patch("regshape.libs.transport.client.resolve_credentials", return_value=(None, None)), + patch("requests.adapters.HTTPAdapter.send", side_effect=_malformed_redirect_adapter), + ): + result = CliRunner().invoke(regshape, [ + "--log-file", str(log_file), "tag", "list", + "-i", f"{REGISTRY}/{REPO}", "--all", "--debug-calls", + ]) + finally: + configure_telemetry(previous) + assert result.exit_code == 1 + assert "Error [" in result.stderr + assert "synthetic-secret" not in result.stderr + assert "synthetic-secret" not in log_file.read_text() + + def test_default_request_parser_errors_remain_unchanged(self) -> None: + """Keep default redirect-enabled callers outside the new sanitization policy.""" + with patch("requests.request", side_effect=ValueError("unchanged")): + with pytest.raises(ValueError, match="unchanged"): + http_request(BASE) + + @pytest.mark.parametrize("status", [301, 302, 303, 307, 308]) + def test_same_origin_redirects_are_validated_and_followed(self, status: int) -> None: + """Continue safe redirects and resolve next links against the final page URL.""" + client = _client([ + _redirect(f"{BASE}/pages//1?n=2", status), + _page(["a"], link='<2?n=2>; rel="next"'), + _page(["b"]), + ]) + assert list_tags_all(client, REPO, page_size=2).tags == ["a", "b"] + assert client.get.call_args_list == [ + call(f"{PATH}?n=2", params=None, allow_redirects=False), + call("/pages//1?n=2", params=None, allow_redirects=False), + call("/pages//2?n=2", params=None, allow_redirects=False), + ] + + def test_missing_redirect_location_is_an_error(self) -> None: + """A redirect without a target cannot be treated as an empty tag page.""" + client = _client([_redirect(None)]) + with pytest.raises(TagError, match="Location"): + list_tags_all(client, REPO) + assert client.get.call_count == 1 + + def test_query_canonicalizing_redirect_is_not_a_completed_page(self) -> None: + """Fetch a valid redirect target even when its logical pagination key matches.""" + client = _client([ + _redirect("?last=0&n=2"), + _page(["1"]), + ]) + assert list_tags_all(client, REPO, page_size=2, last="0").tags == ["1"] + assert client.get.call_args_list == [ + call(f"{PATH}?n=2&last=0", params=None, allow_redirects=False), + call(f"{PATH}?last=0&n=2", params=None, allow_redirects=False), + ] + + def test_canonicalizing_redirect_cycle_is_still_bounded(self) -> None: + """Separate raw redirect-hop identity from normalized completed-page identity.""" + client = _client([ + _redirect("?last=0&n=2"), + _redirect("?n=2&last=0"), + ]) + with pytest.raises(TagError, match="redirect"): + list_tags_all(client, REPO, page_size=2, last="0") + assert client.get.call_count == 2 + + def test_redirect_chains_are_bounded(self) -> None: + """Distinct redirect URLs still have a finite per-page redirect budget.""" + client = _client([_redirect(f"?n=2&hop={index}") for index in range(1, 33)]) + with pytest.raises(TagError, match="redirect limit"): + list_tags_all(client, REPO, page_size=2) + assert client.get.call_count == 31 + + @pytest.mark.parametrize("middleware", [False, True]) + def test_other_transport_requests_retain_redirect_defaults(self, middleware: bool) -> None: + """The new request option must not change unrelated callers by default.""" + with patch("regshape.libs.decorators.call_details.requests.request", return_value=_page([])) as http: + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_middleware=middleware, + )) + client.get(PATH) + assert http.call_args.kwargs.get("allow_redirects", True) is True + + @pytest.mark.parametrize("first_policy", [False, True]) + def test_redirect_policies_do_not_share_cached_responses(self, first_policy: bool) -> None: + """A cached auto-follow result must not bypass an explicit redirect opt-out.""" + first, second = _page(["first"]), _page(["second"]) + with patch( + "regshape.libs.decorators.call_details.requests.request", + side_effect=[first, second], + ) as http: + client = RegistryClient(TransportConfig( + registry=REGISTRY, username="user", password="test-password", + enable_caching=True, + )) + assert client.get(PATH, allow_redirects=first_policy) is first + assert client.get(PATH, allow_redirects=not first_policy) is second + assert client.get(PATH, allow_redirects=first_policy) is first + assert http.call_count == 2 + + def test_parser_errors_hide_credentials_and_raw_exception_context(self) -> None: + """Unicode normalization failures must not echo credential-bearing netlocs.""" + target = f"https://user:secret@{REGISTRY}\uff0fignored/tags" + with pytest.raises(TagError) as error: + _safe_tag_page_url(BASE, target, BASE + PATH) + rendered = "".join(traceback.format_exception( + type(error.value), error.value, error.value.__traceback__, + )) + assert "secret" not in str(error.value) + assert "secret" not in rendered + assert error.value.__suppress_context__ + + def test_cli_does_not_print_credentials_from_malformed_links(self) -> None: + """Only the safe input reference and sanitized failure are sent to stderr.""" + target = f"https://user:secret@{REGISTRY}\uff0fignored/tags" + client = _client([_page(["a"], link=f'<{target}>; rel="next"')]) + with patch("regshape.cli.tag.RegistryClient", return_value=client): + result = CliRunner().invoke(regshape, [ + "tag", "list", "-i", f"{REGISTRY}/{REPO}", "--all", + ]) + assert result.exit_code == 1 + assert "secret" not in result.stderr + assert "pagination URL" in result.stderr + + +class TestListTagsAll: + """Cover OCI link preference, legacy cursors, and explicit failure boundaries.""" + + @pytest.mark.parametrize("tags", [[], ["A", "b"], None]) + def test_default_page_size_and_result_shape(self, tags: object) -> None: + """Return one normalized model with a positive all-pages default.""" + client = _client([_page(tags)]) + result = list_tags_all(client, REPO) + assert result.to_dict() == {"name": REPO, "tags": tags or []} + client.get.assert_called_once_with(f"{PATH}?n=100", params=None, allow_redirects=False) + + @pytest.mark.parametrize("size", [0, -1, True, 1.5, "2"]) + def test_invalid_page_size_fails_before_network(self, size: object) -> None: + """Reject invalid runtime input rather than reporting an incomplete success.""" + client = _client([]) + with pytest.raises(TagError, match="positive page size"): + list_tags_all(client, REPO, page_size=size) # type: ignore[arg-type] + client.get.assert_not_called() + + @pytest.mark.parametrize("cursor", [ + "", "a\nb", "a\n", "a b", "a/b", ".tag", "-tag", "\u00e9", "\ud800", "a" * 129, + ]) + def test_invalid_initial_tag_cursor_fails_before_network(self, cursor: str) -> None: + """Reject malformed tag cursors before URL encoding can fail or any request runs.""" + client = _client([]) + with pytest.raises(TagError, match="valid OCI tag"): + list_tags_all(client, REPO, last=cursor) + client.get.assert_not_called() + + @pytest.mark.parametrize("tag", ["0", "_", "A.Z-rc_1", "a" * 128]) + def test_valid_tag_grammar_preserves_spelling_and_cursors(self, tag: str) -> None: + """Accept valid spelling and the 128-character boundary on every tag surface.""" + assert list_tags_all(_client([_page([tag])]), REPO).tags == [tag] + assert list_tags_all(_client([_page(["z"])]), REPO, last=tag).tags == ["z"] + client = _client([_page([], link=f'; rel="next"'), _page(["z"])]) + assert list_tags_all(client, REPO).tags == ["z"] + + def test_single_page_zero_and_last_only_remain_supported(self) -> None: + """Do not impose all-pages validation on the existing one-page operation.""" + client = _client([_page([]), _page(["b"])]) + assert list_tags(client, REPO, page_size=0).tags == [] + assert list_tags(client, REPO, last="a").tags == ["b"] + assert client.get.call_args_list == [ + call(PATH, params={"n": 0}), + call(PATH, params={"last": "a"}), + ] + + def test_exact_multiple_without_links_uses_final_empty_page(self) -> None: + """Use exclusive tag values, including numeric tags, rather than offsets.""" + client = _client([_page(["10", "20"]), _page(["A", "b"]), _page([])]) + result = list_tags_all(client, REPO, page_size=2, last="0") + assert result.tags == ["10", "20", "A", "b"] + assert client.get.call_args_list == [ + call(f"{PATH}?n=2&last=0", params=None, allow_redirects=False), + call(f"{PATH}?n=2&last=20", params=None, allow_redirects=False), + call(f"{PATH}?n=2&last=b", params=None, allow_redirects=False), + ] + + @pytest.mark.parametrize("target", [ + f"{BASE}{PATH}?n=2&last=a", + f"https://REGISTRY.EXAMPLE.COM:443{PATH}?n=2&last=a", + f"{PATH}?n=2&last=a", + "list?n=2&last=a", + "?n=2&last=a", + ]) + @pytest.mark.parametrize("first_tags", [[], ["a"]]) + def test_next_link_wins_on_short_or_empty_page( + self, target: str, first_tags: list[str] + ) -> None: + """Follow relative/absolute next links even when page length is below n.""" + client = _client([ + _page(first_tags, link=f'<{target}>; rel="next"'), + _page(["b"]), + ]) + client.last_response = _page([], link='; rel="next"') + result = list_tags_all(client, REPO, page_size=2) + assert result.tags == first_tags + ["b"] + assert client.get.call_args_list == [ + call(f"{PATH}?n=2", params=None, allow_redirects=False), + call(f"{PATH}?n=2&last=a", params=None, allow_redirects=False), + ] + + def test_response_url_and_linked_query_are_authoritative(self) -> None: + """Resolve links against the actual response and preserve opaque query values.""" + target = "2?n=2&cursor=a,b;c==d&token=x%2Fy" + header = f'; rel="prev", <{target}>; title="a, ; c"; rel="next"' + client = _client([ + _page(["a"], link=header, url=f"{BASE}/pages/1?n=5"), + _page(["b"]), + ]) + assert list_tags_all(client, REPO, page_size=5).tags == ["a", "b"] + assert client.get.call_args_list[1] == call( + "/pages/2?n=2&cursor=a,b;c==d&token=x%2Fy", params=None, allow_redirects=False, + ) + + @pytest.mark.parametrize(("target", "expected_path"), [ + ("pages/token//part?n=2", f"/v2/{REPO}/tags/pages/token//part?n=2"), + ("pages/token//../part?n=2", f"/v2/{REPO}/tags/pages/token/part?n=2"), + ("./pages//part/../next?n=2", f"/v2/{REPO}/tags/pages//next?n=2"), + ]) + def test_relative_links_preserve_empty_path_segments( + self, target: str, expected_path: str + ) -> None: + """Resolve relative paths without discarding opaque empty URI segments.""" + client = _client([ + _page(["a"], link=f'<{target}>; rel="next"'), + _page(["b"]), + ]) + assert list_tags_all(client, REPO, page_size=2).tags == ["a", "b"] + assert client.get.call_args == call(expected_path, params=None, allow_redirects=False) + + def test_empty_path_segments_in_response_base_are_preserved(self) -> None: + """Retain empty segments inherited from the actual response URL.""" + client = _client([ + _page(["a"], link='<2?n=2>; rel="next"', url=f"{BASE}/pages//1?n=2"), + _page(["b"]), + ]) + assert list_tags_all(client, REPO, page_size=2).tags == ["a", "b"] + assert client.get.call_args == call("/pages//2?n=2", params=None, allow_redirects=False) + + def test_linkless_fallback_uses_changed_page_size_and_keeps_query(self) -> None: + """A linked n=2 page must not be mistaken for a short initial n=5 page.""" + client = _client([ + _page(["a"], link='; rel="next"'), + _page(["b", "c"]), + _page(["d"]), + ]) + assert list_tags_all(client, REPO, page_size=5).tags == ["a", "b", "c", "d"] + assert client.get.call_args_list == [ + call(f"{PATH}?n=5", params=None, allow_redirects=False), + call(f"{PATH}?n=2&cursor=next", params=None, allow_redirects=False), + call(f"{PATH}?n=2&cursor=next&last=c", params=None, allow_redirects=False), + ] + + def test_cursor_fallback_preserves_raw_opaque_fields(self) -> None: + """Do not decode binary cursors, normalize escapes, or rewrite opaque flags.""" + query = "n=%32&cursor=%FF&opaque=%2f%20%3d&flag&dup=one&dup=two" + client = _client([ + _page(["a"], link=f'; rel="next"'), + _page(["b", "c"]), + _page(["d"]), + ]) + assert list_tags_all(client, REPO, page_size=2).tags == ["a", "b", "c", "d"] + assert client.get.call_args == call(f"{PATH}?{query}&last=c", params=None, allow_redirects=False) + + def test_distinct_binary_cursors_are_not_the_same_page(self) -> None: + """Opaque byte sequences must not collapse into one replacement character.""" + client = _client([ + _page([], link='; rel="next"'), + _page([], link='; rel="next"'), + _page(["a"]), + ]) + assert list_tags_all(client, REPO, page_size=2).tags == ["a"] + assert client.get.call_args_list == [ + call(f"{PATH}?n=2", params=None, allow_redirects=False), + call(f"{PATH}?n=2&cursor=%FF", params=None, allow_redirects=False), + call(f"{PATH}?n=2&cursor=%FE", params=None, allow_redirects=False), + ] + + def test_link_may_omit_n_when_using_last(self) -> None: + """Without a requested limit or next link, the linked response is complete.""" + client = _client([ + _page(["a"], link='; rel="next"'), + _page(["b", "c", "d"]), + ]) + assert list_tags_all(client, REPO, page_size=1).tags == ["a", "b", "c", "d"] + assert client.get.call_count == 2 + assert client.get.call_args == call(f"{PATH}?last=a", params=None, allow_redirects=False) + + def test_absolute_link_with_empty_path_uses_registry_root(self) -> None: + """Normalize an absolute origin-only target to an HTTP path, not a bare query.""" + client = _client([ + _page(["a"], link=f'<{BASE}?n=2&last=a>; rel="next"'), + _page(["b"]), + ]) + assert list_tags_all(client, REPO, page_size=2).tags == ["a", "b"] + assert client.get.call_args == call("/?n=2&last=a", params=None, allow_redirects=False) + + @pytest.mark.parametrize("query", [ + "n=0", "n=-1", "n=oops", "n=1_0", "n=+2", "n=2.0", + "n=1&n=2", "last=a&last=b", "last=", "n=%FF", "last=%FF", + "last=a%0Ab", "last=a%0A", "last=a+b", "last=a%2Fb", "last=.tag", + "last=-tag", "last=%C3%A9", "last=%ED%A0%80", f"last={'a' * 129}", + ]) + def test_bad_linked_parameters_fail_before_following(self, query: str) -> None: + """Never fetch a linked page whose pagination state is invalid or ambiguous.""" + client = _client([_page(["a"], link=f'; rel="next"')]) + with pytest.raises(TagError, match="pagination"): + list_tags_all(client, REPO, page_size=2) + assert client.get.call_count == 1 + + @pytest.mark.parametrize("target", [ + "https://other.example/tags", + "//other.example/tags", + f"http://{REGISTRY}{PATH}", + f"https://{REGISTRY}:444{PATH}", + f"https://{REGISTRY}:0{PATH}", + f"https://user:secret@{REGISTRY}{PATH}", + f"https://{REGISTRY}:invalid{PATH}", + "https://[broken", + "ftp://registry.example.com/tags", + "/\\other.example/tags", + "/tags?cursor=raw space", + ]) + def test_unsafe_next_links_are_rejected_before_credentials_can_be_sent( + self, target: str + ) -> None: + """Reject cross-origin, downgrade, userinfo, and malformed destinations.""" + client = _client([_page(["a"], link=f'<{target}>; rel="next"')]) + with pytest.raises(TagError, match="pagination URL") as error: + list_tags_all(client, REPO) + assert "secret" not in str(error.value) + assert client.get.call_count == 1 + + def test_explicit_insecure_registry_can_follow_same_origin_http_links(self) -> None: + """HTTP is allowed only when it is the configured registry origin.""" + client = _client([ + _page(["a"], link=f'; rel="next"'), + _page(["b"]), + ], insecure=True) + assert list_tags_all(client, REPO, page_size=2).tags == ["a", "b"] + + def test_url_cycle_is_bounded_even_for_empty_pages(self) -> None: + """Stop a repeated next-page URL without requesting it a second time.""" + client = _client([ + _page([], link='; rel="next"'), + _page([], link='; rel="next"'), + ]) + with pytest.raises(TagError, match="repeated a page"): + list_tags_all(client, REPO, page_size=2) + assert client.get.call_count == 2 + + def test_duplicate_pages_with_changing_links_are_not_progress(self) -> None: + """Changing an opaque URL must not hide repeatedly returned tags.""" + client = _client([ + _page(["a"], link='; rel="next"'), + _page(["a"], link='; rel="next"'), + ]) + with pytest.raises(TagError, match="no progress"): + list_tags_all(client, REPO, page_size=2) + assert client.get.call_count == 2 + + def test_empty_linked_pages_stop_at_ten_despite_changing_links(self) -> None: + """Bound empty traversal even when opaque URLs hide repeated tag cursors.""" + client = _client(_empty_linked_pages(11)) + with pytest.raises(TagError, match="10 consecutive empty linked pages"): + list_tags_all(client, REPO, page_size=2) + assert client.get.call_count == 10 + + def test_nine_empty_linked_pages_can_reach_tags(self) -> None: + """Do not reject valid continuation before the empty-page threshold.""" + client = _client([*_empty_linked_pages(9), _page(["c"])]) + assert list_tags_all(client, REPO, page_size=2).tags == ["c"] + assert client.get.call_count == 10 + + def test_tenth_empty_page_without_next_link_finishes(self) -> None: + """An unlinked empty page completes normally instead of exhausting the budget.""" + client = _client([*_empty_linked_pages(9), _page([])]) + assert list_tags_all(client, REPO, page_size=2).tags == [] + assert client.get.call_count == 10 + + @pytest.mark.parametrize("linked_progress", [False, True]) + def test_tags_reset_the_empty_linked_page_counter(self, linked_progress: bool) -> None: + """Reset on received tags for both linked and cursor-only continuation.""" + client = _client([ + *_empty_linked_pages(9), + _page( + ["c", "d"], + link='; rel="next"' if linked_progress else None, + ), + *_empty_linked_pages(9, start=10), + _page(["e"]), + ]) + assert list_tags_all(client, REPO, page_size=2).tags == ["c", "d", "e"] + assert client.get.call_count == 20 + + def test_redirect_is_not_an_empty_linked_page(self) -> None: + """Count tag pages, not redirect responses, toward the empty-page limit.""" + client = _client([ + *_empty_linked_pages(9), + _redirect("?n=2&nonce=redirect"), + _page(["c"]), + ]) + assert list_tags_all(client, REPO, page_size=2).tags == ["c"] + assert client.get.call_count == 11 + + def test_redirect_does_not_reset_empty_linked_page_counter(self) -> None: + """Keep consecutive empty-page accounting across validated redirects.""" + client = _client([ + *_empty_linked_pages(9), + _redirect("?n=2&nonce=redirect"), + *_empty_linked_pages(1, start=9), + _page(["c"]), + ]) + with pytest.raises(TagError, match="10 consecutive empty linked pages"): + list_tags_all(client, REPO, page_size=2) + assert client.get.call_count == 11 + + @pytest.mark.parametrize("pages", [ + [_page(["a", "a"])], + [_page(["a", "b"], link='; rel="next"'), _page(["b", "c"])], + ]) + def test_repeated_tags_are_not_returned_as_a_successful_list( + self, pages: list[requests.Response] + ) -> None: + """Reject within-page duplicates and partial page overlap without deduplicating.""" + client = _client(pages) + with pytest.raises(TagError, match="repeated a tag"): + list_tags_all(client, REPO, page_size=3) + assert client.get.call_count == len(pages) + + def test_cursor_does_not_repeat_the_initial_cursor(self) -> None: + """Reject a full page that would continue with an already used last value.""" + client = _client([_page(["new", "old"])]) + with pytest.raises(TagError, match="repeated a cursor"): + list_tags_all(client, REPO, page_size=2, last="old") + assert client.get.call_count == 1 + + @pytest.mark.parametrize(("last", "pages"), [ + ("a", [_page(["a", "b"])]), + (None, [_page([], link='; rel="next"'), _page(["a", "b"])]), + ("a", [_redirect("?n=3"), _page(["a", "b"])]), + (None, [ + _page([], link='; rel="next"'), + _redirect("?n=3&cursor=next"), + _page(["a", "b"]), + ]), + ]) + def test_short_pages_cannot_include_an_exclusive_cursor( + self, last: str | None, pages: list[requests.Response] + ) -> None: + """Retain initial and linked exclusive boundaries, including across redirects.""" + client = _client(pages) + with pytest.raises(TagError, match="repeated a cursor"): + list_tags_all(client, REPO, page_size=3, last=last) + assert client.get.call_count == len(pages) + + @pytest.mark.parametrize("bad_page", [ + _page([123]), + _page([""]), + *[_page([tag]) for tag in [ + "a\nb", "a\n", "a\rb", "a\tb", "a\x00b", "a b", "a/b", + ".tag", "-tag", "\u00e9", "\ud800", "a" * 129, + ]], + _page("not a list"), + _page([], namespace="other/repository"), + ]) + def test_malformed_page_cannot_be_aggregated(self, bad_page: requests.Response) -> None: + """Reject bad tag values, list shapes, or repository identity.""" + client = _client([bad_page]) + with pytest.raises(TagError): + list_tags_all(client, REPO) + assert client.get.call_count == 1 + + def test_malformed_json_is_a_domain_error(self) -> None: + """Use the existing tag parser error rather than leaking a decoding failure.""" + response = _page([]) + response._content = b"not-json" + with pytest.raises(TagError, match="parse"): + list_tags_all(_client([response]), REPO) + + @pytest.mark.parametrize(("status", "error"), [(401, AuthError), (404, TagError), (500, TagError)]) + def test_later_page_registry_errors_propagate(self, status: int, error: type[Exception]) -> None: + """A failed later page must not become a successful partial TagList.""" + client = _client([_page(["a"]), _page([], status=status)]) + with pytest.raises(error): + list_tags_all(client, REPO, page_size=1) + assert client.get.call_count == 2 + + def test_later_page_transport_error_propagates(self) -> None: + """Keep transport failures explicit after a successful first page.""" + client = _client([_page(["a"]), requests.exceptions.Timeout("timed out")]) + with pytest.raises(requests.exceptions.Timeout, match="timed out"): + list_tags_all(client, REPO, page_size=1) + assert client.get.call_count == 2 + + @pytest.mark.parametrize("middleware", [False, True]) + def test_real_transport_supports_both_client_modes(self, middleware: bool) -> None: + """Exercise requests through RegistryClient, including its legacy URL handling.""" + pages = [_page(["a"], link='; rel="next"'), _page(["b"])] + with ( + patch("regshape.libs.transport.client.resolve_credentials", return_value=(None, None)), + patch("regshape.libs.decorators.call_details.requests.request", side_effect=pages) as http, + ): + client = RegistryClient(TransportConfig( + registry=REGISTRY, enable_middleware=middleware, + )) + assert list_tags_all(client, REPO, page_size=2).tags == ["a", "b"] + assert [entry.args[:2] for entry in http.call_args_list] == [ + ("GET", f"{BASE}{PATH}?n=2"), + ("GET", f"{BASE}{PATH}?n=2&last=a"), + ] + + +class TestTagURLResolution: + """Cover RFC 3986 reference resolution without opaque-path normalization.""" + + @pytest.mark.parametrize(("target", "expected"), [ + ("g", "/b/c/g"), + ("./g", "/b/c/g"), + ("g/", "/b/c/g/"), + ("/g", "/g"), + ("?y", "/b/c/d;p?y"), + ("g?y", "/b/c/g?y"), + ("#s", "/b/c/d;p?q"), + ("g?y#s", "/b/c/g?y"), + (";x", "/b/c/;x"), + ("", "/b/c/d;p?q"), + (".", "/b/c/"), + ("..", "/b/"), + ("../g", "/b/g"), + ("../..", "/"), + ("../../g", "/g"), + ("../../../g", "/g"), + ("/./g", "/g"), + ("/../g", "/g"), + ("g/./h", "/b/c/g/h"), + ("g/../h", "/b/c/h"), + ("g?y/./x", "/b/c/g?y/./x"), + ("g//h", "/b/c/g//h"), + ("g//../h", "/b/c/g/h"), + ("?", "/b/c/d;p"), + ]) + def test_reference_resolution(self, target: str, expected: str) -> None: + """Resolve dot segments and query inheritance while retaining empty segments.""" + assert _safe_tag_page_url(BASE, target, f"{BASE}/b/c/d;p?q") == BASE + expected + + def test_unicode_authorities_cannot_alias_an_ascii_registry(self) -> None: + """Reject a Unicode spelling that Requests sends to a different DNS name.""" + origin = "https://fass.example" + target = "https://fa\u00df.example/challenge" + assert requests.Request("GET", target).prepare().url == "https://xn--fa-hia.example/challenge" + with pytest.raises(TagError): + _safe_tag_page_url(origin, target, origin) + + def test_ascii_punycode_authorities_remain_supported(self) -> None: + """Internationalized domains remain usable through their unambiguous ASCII form.""" + origin = "https://xn--fa-hia.example" + target = origin + "/challenge" + assert _safe_tag_page_url(origin, target, origin) == target + + +class TestNextTagLink: + """Check RFC link syntax beyond comma-splitting and substring matching.""" + + @pytest.mark.parametrize(("header", "expected"), [ + ("", None), + ('; rel="prev"', None), + ('; rel=next', "next-page"), + ('; rel="prev next"', "next-page"), + ('; rel="NEXT"', "next-page"), + ('; rel=prev; rel=next', None), + ('; extension; title="comma, and ; semicolon"; rel=next', "next-page"), + ('; title="escaped \\"quote\\""; rel=next', "next-page?a=b,c;d=e"), + ('; title="contains rel=next", ; rel=next', "next-page"), + ('; rel=next, ; rel=next', "same"), + ]) + def test_selects_the_next_relation(self, header: str, expected: str | None) -> None: + """Honor relation tokens and quoted parameters without altering URI contents.""" + assert _next_tag_link(header) == expected + + @pytest.mark.parametrize("header", [ + "broken", + "; rel=", + '; rel=""', + '; rel=" "', + '; rel="unterminated', + "; rel=next trailing", + "; rel", + '; anchor="/elsewhere"; rel=next', + "; rel=next, ; rel=next", + ]) + def test_malformed_or_ambiguous_continuations_fail(self, header: str) -> None: + """Do not silently treat an unusable pagination header as end of results.""" + with pytest.raises(TagError): + _next_tag_link(header) + + +class TestTagPaginationCLI: + """Preserve CLI formatting and fail before emitting incomplete results.""" + + @pytest.mark.parametrize("size", ["0", "-2"]) + def test_invalid_all_size_is_a_usage_error_before_client_creation(self, size: str) -> None: + """Reject invalid all-pages sizes before credential lookup or requests.""" + with patch("regshape.cli.tag.RegistryClient") as factory: + result = CliRunner().invoke(regshape, [ + "tag", "list", "-i", f"{REGISTRY}/{REPO}", "--all", "--n", size, + ]) + assert result.exit_code == 2 + assert "--n" in result.stderr + factory.assert_not_called() + + @pytest.mark.parametrize("as_json", [False, True]) + @pytest.mark.parametrize("write_file", [False, True]) + def test_all_preserves_options_and_output_shape( + self, tmp_path: Path, as_json: bool, write_file: bool + ) -> None: + """Forward page size/cursor and serialize the complete result exactly once.""" + output = tmp_path / "tags.txt" + tags = TagList(namespace=REPO, tags=["B", "c"]) + args = [ + "tag", "list", "-i", f"{REGISTRY}/{REPO}:ignored", + "--all", "--n", "2", "--last", "a", + ] + if as_json: + args.append("--json") + if write_file: + args.extend(["--output", str(output)]) + with ( + patch("regshape.cli.tag.RegistryClient") as factory, + patch("regshape.cli.tag.list_tags_all", return_value=tags) as fetch, + patch("regshape.cli.tag.list_tags") as single, + ): + result = CliRunner().invoke(regshape, args) + assert result.exit_code == 0, result.output + fetch.assert_called_once_with( + client=factory.return_value, repo=REPO, page_size=2, last="a", + ) + single.assert_not_called() + content = output.read_text() if write_file else result.stdout + if write_file: + assert result.stdout == "" + if as_json: + assert json.loads(content) == tags.to_dict() + else: + assert content == "B\nc\n" + + def test_single_page_zero_remains_a_cli_option(self) -> None: + """Keep n=0 available when automatic traversal is not requested.""" + with ( + patch("regshape.cli.tag.RegistryClient"), + patch("regshape.cli.tag.list_tags", return_value=TagList(REPO, [])) as single, + patch("regshape.cli.tag.list_tags_all") as all_pages, + ): + result = CliRunner().invoke(regshape, [ + "tag", "list", "-i", f"{REGISTRY}/{REPO}", "--n", "0", + ]) + assert result.exit_code == 0 + assert single.call_args.kwargs["page_size"] == 0 + all_pages.assert_not_called() + + @pytest.mark.parametrize("insecure", [False, True]) + def test_complete_cli_transport_flow_and_telemetry(self, insecure: bool) -> None: + """Traverse through the real library/client and preserve telemetry separation.""" + args = ["--insecure"] if insecure else [] + args.extend([ + "tag", "list", "-i", f"{REGISTRY}/{REPO}", "--all", "--n", "1", + "--json", "--debug-calls", "--time-scenarios", + ]) + pages = [_page(["a"]), _page(["b"]), _page([])] + with ( + patch("regshape.libs.transport.client.resolve_credentials", return_value=(None, None)), + patch("regshape.libs.decorators.call_details.requests.request", side_effect=pages) as http, + ): + result = CliRunner().invoke(regshape, args) + assert result.exit_code == 0, result.output + assert json.loads(result.stdout) == {"name": REPO, "tags": ["a", "b"]} + assert "tag list" in result.stderr + assert "GET" in result.stderr + scheme = "http" if insecure else "https" + assert [entry.args[1] for entry in http.call_args_list] == [ + f"{scheme}://{REGISTRY}{PATH}?n=1", + f"{scheme}://{REGISTRY}{PATH}?n=1&last=a", + f"{scheme}://{REGISTRY}{PATH}?n=1&last=b", + ] + + def test_later_page_failure_preserves_existing_output(self, tmp_path: Path) -> None: + """Do not print success or overwrite a file with only the first page.""" + output = tmp_path / "tags.json" + output.write_text("unchanged\n") + with ( + patch("regshape.libs.transport.client.resolve_credentials", return_value=(None, None)), + patch("regshape.libs.decorators.call_details.requests.request", + side_effect=[_page(["a"]), _page([], status=500)]), + ): + result = CliRunner().invoke(regshape, [ + "tag", "list", "-i", f"{REGISTRY}/{REPO}", "--all", "--n", "1", + "--json", "--output", str(output), + ]) + assert result.exit_code == 1 + assert result.stdout == "" + assert "Registry error" in result.stderr + assert output.read_text() == "unchanged\n" + + def test_overlapping_pages_preserve_existing_output(self, tmp_path: Path) -> None: + """Do not write a duplicated or partially deduplicated all-pages result.""" + output = tmp_path / "tags.json" + output.write_text("unchanged\n") + client = _client([ + _page(["a", "b"], link='; rel="next"'), + _page(["b", "c"]), + ]) + with patch("regshape.cli.tag.RegistryClient", return_value=client): + result = CliRunner().invoke(regshape, [ + "tag", "list", "-i", f"{REGISTRY}/{REPO}", "--all", + "--n", "3", "--json", "--output", str(output), + ]) + assert result.exit_code == 1 + assert "repeated a tag" in result.stderr + assert result.stdout == "" + assert output.read_text() == "unchanged\n" + + @pytest.mark.parametrize("as_json", [False, True]) + @pytest.mark.parametrize("destination", ["stdout", "existing-file", "new-file"]) + def test_empty_linked_page_limit_never_emits_partial_results( + self, tmp_path: Path, as_json: bool, destination: str + ) -> None: + """Surface the real traversal limit without printing or writing partial tags.""" + output = tmp_path / "tags.txt" + if destination == "existing-file": + output.write_text("unchanged\n") + args = ["tag", "list", "-i", f"{REGISTRY}/{REPO}", "--all", "--n", "1"] + if as_json: + args.append("--json") + if destination != "stdout": + args.extend(["--output", str(output)]) + with ( + patch("regshape.libs.transport.client.resolve_credentials", return_value=(None, None)), + patch( + "regshape.libs.decorators.call_details.requests.request", + side_effect=[_page(["c"]), *_empty_linked_pages(11)], + ) as http, + ): + result = CliRunner().invoke(regshape, args) + assert result.exit_code == 1 + assert "10 consecutive empty linked pages" in result.stderr + assert result.stdout == "" + assert http.call_count == 11 + if destination == "existing-file": + assert output.read_text() == "unchanged\n" + else: + assert not output.exists() + + @pytest.mark.parametrize("tag", ["a\nb", "\ud800"]) + @pytest.mark.parametrize("as_json", [False, True]) + @pytest.mark.parametrize("write_file", [False, True]) + def test_malformed_tags_fail_without_emitting_partial_output( + self, tmp_path: Path, tag: str, as_json: bool, write_file: bool + ) -> None: + """Reject malformed later-page tags before rendering or cursor encoding.""" + output = tmp_path / "tags.txt" + output.write_text("unchanged\n") + args = ["tag", "list", "-i", f"{REGISTRY}/{REPO}", "--all", "--n", "1"] + if as_json: + args.append("--json") + if write_file: + args.extend(["--output", str(output)]) + client = _client([_page(["A"]), _page([tag])]) + with patch("regshape.cli.tag.RegistryClient", return_value=client): + result = CliRunner().invoke(regshape, args) + assert result.exit_code == 1 + assert "invalid tag value" in result.stderr + assert result.stdout == "" + assert output.read_text() == "unchanged\n" + assert client.get.call_count == 2 + + def test_help_explains_all_pages_and_default(self) -> None: + """Make the opt-in behavior and positive all-pages default discoverable.""" + result = CliRunner().invoke(regshape, ["tag", "list", "--help"]) + assert result.exit_code == 0 + assert "--all" in result.output + assert "100" in result.output + assert "one page only" in " ".join(result.output.split()) diff --git a/src/regshape/tests/test_transport_models.py b/src/regshape/tests/test_transport_models.py index 0f193b2..adb12d8 100644 --- a/src/regshape/tests/test_transport_models.py +++ b/src/regshape/tests/test_transport_models.py @@ -11,6 +11,7 @@ class TestRegistryRequest: def test_valid_request_creation(self): + """Default requests retain automatic redirects for compatibility.""" req = RegistryRequest( method="GET", url="https://registry.example.com/v2/repo/manifests/tag", @@ -21,6 +22,7 @@ def test_valid_request_creation(self): assert req.headers["Accept"] == "application/vnd.oci.image.manifest.v1+json" assert req.body is None assert req.stream is False + assert req.allow_redirects is True def test_request_with_body_and_params(self): body = b'{"some": "json"}'