Add URL rewrite support for schema downloads

CHANGELOG.rst

@@ -10,6 +10,8 @@ Unreleased
 
 .. vendor-insert-here
 
+- Add a repeatable ``--url-rewrite SOURCE_PREFIX TARGET_PREFIX`` option for
+  downloading remote schemas and references through mirrors. (:issue:`680`)
 - Update vendored schemas: bitbucket-pipelines, mergify, renovate (2026-08-16)
 
 0.38.0

docs/usage.rst

@@ -141,6 +141,18 @@ The following options control caching behaviors.
      - Description
    * - ``--no-cache``
      - Disable caching.
+   * - ``--url-rewrite SOURCE_PREFIX TARGET_PREFIX``
+     - Download matching schema URLs from a different HTTP(S) location. May be
+       specified multiple times; the longest matching source prefix wins.
+
+URL rewrites make remote schemas available through a mirror without changing their
+logical retrieval URI. They apply to both the initial ``--schemafile`` URL and remote
+``$ref`` URLs. For example::
+
+    check-jsonschema \
+      --schemafile https://www.schemastore.org/github-workflow.json \
+      --url-rewrite https://www.schemastore.org/ https://schemas.example/mirror/ \
+      .github/workflows/ci.yml
 
 "format" Validation Options
 ---------------------------

src/check_jsonschema/cachedownloader.py

@@ -121,9 +121,24 @@ class FailedDownloadError(Exception):
 
 
 class CacheDownloader:
-    def __init__(self, cache_dir: str, *, disable_cache: bool = False) -> None:
+    def __init__(
+        self,
+        cache_dir: str,
+        *,
+        disable_cache: bool = False,
+        url_rewrites: tuple[tuple[str, str], ...] = (),
+    ) -> None:
         self._cache_dir = _resolve_cache_dir(cache_dir)
         self._disable_cache = disable_cache
+        self._url_rewrites = url_rewrites
+
+    def _rewrite_url(self, file_url: str) -> str:
+        matches = (rule for rule in self._url_rewrites if file_url.startswith(rule[0]))
+        rule = max(matches, key=lambda item: len(item[0]), default=None)
+        if rule is None:
+            return file_url
+        source, replacement = rule
+        return f"{replacement}{file_url[len(source) :]}"
 
     def _download(
         self,
@@ -144,7 +159,9 @@ def check_response_for_download(r: requests.Response) -> bool:
             # we now know it's not a hit, so validate the content (forces download)
             return response_ok(r)
 
-        response = _get_request(file_url, response_ok=check_response_for_download)
+        response = _get_request(
+            self._rewrite_url(file_url), response_ok=check_response_for_download
+        )
         # check to see if we have a file which matches the connection
         # only download if we do not (cache miss, vs hit)
         if not _cache_hit(dest, response):
@@ -161,7 +178,9 @@ def open(
     ) -> t.Iterator[t.IO[bytes]]:
         if (not self._cache_dir) or self._disable_cache:
             yield io.BytesIO(
-                _get_request(file_url, response_ok=validate_response).content
+                _get_request(
+                    self._rewrite_url(file_url), response_ok=validate_response
+                ).content
             )
         else:
             with open(

src/check_jsonschema/cli/main_command.py

@@ -3,6 +3,7 @@
 import os
 import textwrap
 import typing as t
+import urllib.parse
 
 import click
 import jsonschema
@@ -57,6 +58,22 @@ def pretty_helptext_list(values: list[str] | tuple[str, ...]) -> str:
     )
 
 
+def validate_url_rewrites(
+    ctx: click.Context,
+    param: click.Parameter,
+    value: tuple[tuple[str, str], ...],
+) -> tuple[tuple[str, str], ...]:
+    del ctx
+    for source, target in value:
+        for url in (source, target):
+            parsed = urllib.parse.urlsplit(url)
+            if parsed.scheme not in ("http", "https") or not parsed.netloc:
+                raise click.BadParameter(
+                    "both prefixes must be absolute HTTP(S) URLs", param=param
+                )
+    return value
+
+
 @click.command(
     "check-jsonschema",
     help="""\
@@ -125,6 +142,17 @@ def pretty_helptext_list(values: list[str] | tuple[str, ...]) -> str:
     is_flag=True,
     help="Disable schema caching. Always download remote schemas.",
 )
+@click.option(
+    "--url-rewrite",
+    type=(str, str),
+    multiple=True,
+    callback=validate_url_rewrites,
+    metavar="SOURCE_PREFIX TARGET_PREFIX",
+    help=(
+        "Rewrite matching HTTP(S) schema URLs before downloading. May be repeated; "
+        "the longest matching source prefix wins."
+    ),
+)
 @click.option(
     "--cache-filename", help="Deprecated. This option no longer has any effect."
 )
@@ -242,6 +270,7 @@ def main(
     base_uri: str | None,
     check_metaschema: bool,
     no_cache: bool,
+    url_rewrite: tuple[tuple[str, str], ...],
     cache_filename: str | None,
     disable_formats: tuple[list[str], ...],
     format_regex: t.Literal["python", "nonunicode", "default"] | None,
@@ -276,6 +305,7 @@ def main(
         args.disable_formats = normalized_disable_formats
 
     args.disable_cache = no_cache
+    args.url_rewrites = url_rewrite
     args.default_filetype = default_filetype
     args.force_filetype = force_filetype
     args.fill_defaults = fill_defaults
@@ -301,14 +331,19 @@ def build_schema_loader(args: ParseResult) -> SchemaLoaderBase:
         return MetaSchemaLoader(base_uri=args.base_uri)
     elif args.schema_mode == SchemaLoadingMode.builtin:
         assert args.schema_path is not None
-        return BuiltinSchemaLoader(args.schema_path, base_uri=args.base_uri)
+        return BuiltinSchemaLoader(
+            args.schema_path,
+            base_uri=args.base_uri,
+            url_rewrites=args.url_rewrites,
+        )
     elif args.schema_mode == SchemaLoadingMode.filepath:
         assert args.schema_path is not None
         return SchemaLoader(
             args.schema_path,
             disable_cache=args.disable_cache,
             base_uri=args.base_uri,
             validator_class=args.validator_class,
+            url_rewrites=args.url_rewrites,
         )
     else:
         raise NotImplementedError("no valid schema option provided")

src/check_jsonschema/cli/parse_result.py

@@ -27,6 +27,7 @@ def __init__(self) -> None:
         # cache controls
         self.disable_cache: bool = False
         self.cache_filename: str | None = None
+        self.url_rewrites: tuple[tuple[str, str], ...] = ()
         # filetype detection (JSON, YAML, TOML, etc)
         self.default_filetype: str = "json"
         self.force_filetype: str | None = None

src/check_jsonschema/schema_loader/main.py

@@ -75,6 +75,7 @@ def get_validator(
 class SchemaLoader(SchemaLoaderBase):
     validator_class: type[jsonschema.protocols.Validator] | None = None
     disable_cache: bool = True
+    url_rewrites: tuple[tuple[str, str], ...] = ()
 
     def __init__(
         self,
@@ -83,12 +84,14 @@ def __init__(
         base_uri: str | None = None,
         validator_class: type[jsonschema.protocols.Validator] | None = None,
         disable_cache: bool = True,
+        url_rewrites: tuple[tuple[str, str], ...] = (),
     ) -> None:
         # record input parameters (these are not to be modified)
         self.schemafile = schemafile
         self.disable_cache = disable_cache
         self.base_uri = base_uri
         self.validator_class = validator_class
+        self.url_rewrites = url_rewrites
 
         # if the schema location is a URL, which may include a file:// URL, parse it
         self.url_info = None
@@ -119,7 +122,9 @@ def _get_schema_reader(
             return LocalSchemaReader(self.schemafile)
 
         if self.url_info.scheme in ("http", "https"):
-            return HttpSchemaReader(self.schemafile, self.disable_cache)
+            return HttpSchemaReader(
+                self.schemafile, self.disable_cache, self.url_rewrites
+            )
         else:
             raise UnsupportedUrlScheme(
                 "check-jsonschema only supports http, https, and local files. "
@@ -162,7 +167,11 @@ def _get_validator(
         # reference resolution
         # with support for YAML, TOML, and other formats from the parsers
         reference_registry = make_reference_registry(
-            self._parsers, retrieval_uri, schema, self.disable_cache
+            self._parsers,
+            retrieval_uri,
+            schema,
+            self.disable_cache,
+            self.url_rewrites,
         )
 
         if self.validator_class is None:
@@ -241,9 +250,16 @@ def _dialect_of_schema(schema: dict[str, t.Any] | bool) -> str | None:
 
 
 class BuiltinSchemaLoader(SchemaLoader):
-    def __init__(self, schema_name: str, *, base_uri: str | None = None) -> None:
+    def __init__(
+        self,
+        schema_name: str,
+        *,
+        base_uri: str | None = None,
+        url_rewrites: tuple[tuple[str, str], ...] = (),
+    ) -> None:
         self.schema_name = schema_name
         self.base_uri = base_uri
+        self.url_rewrites = url_rewrites
         self._parsers = ParserSet()
 
     def get_schema_retrieval_uri(self) -> str | None:

src/check_jsonschema/schema_loader/readers.py

@@ -74,12 +74,13 @@ def __init__(
         self,
         url: str,
         disable_cache: bool,
+        url_rewrites: tuple[tuple[str, str], ...] = (),
     ) -> None:
         self.url = url
         self.parsers = ParserSet()
-        self.downloader = CacheDownloader("schemas", disable_cache=disable_cache).bind(
-            url, validation_callback=self._parse
-        )
+        self.downloader = CacheDownloader(
+            "schemas", disable_cache=disable_cache, url_rewrites=url_rewrites
+        ).bind(url, validation_callback=self._parse)
         self._parsed_schema: dict | _UnsetType = _UNSET
 
     def _parse(self, schema_bytes: bytes) -> t.Any:

src/check_jsonschema/schema_loader/resolver.py

@@ -12,7 +12,11 @@
 
 
 def make_reference_registry(
-    parsers: ParserSet, retrieval_uri: str | None, schema: dict, disable_cache: bool
+    parsers: ParserSet,
+    retrieval_uri: str | None,
+    schema: dict,
+    disable_cache: bool,
+    url_rewrites: tuple[tuple[str, str], ...] = (),
 ) -> referencing.Registry:
     id_attribute_: t.Any = schema.get("$id")
     if isinstance(id_attribute_, str):
@@ -27,7 +31,7 @@ def make_reference_registry(
     # argument to its implicit initializer
     registry: referencing.Registry = referencing.Registry(  # type: ignore[call-arg]
         retrieve=create_retrieve_callable(
-            parsers, retrieval_uri, id_attribute, disable_cache
+            parsers, retrieval_uri, id_attribute, disable_cache, url_rewrites
         )
     )
 
@@ -44,13 +48,16 @@ def create_retrieve_callable(
     retrieval_uri: str | None,
     id_attribute: str | None,
     disable_cache: bool,
+    url_rewrites: tuple[tuple[str, str], ...] = (),
 ) -> t.Callable[[str], referencing.Resource[Schema]]:
     base_uri = id_attribute
     if base_uri is None:
         base_uri = retrieval_uri
 
     cache = ResourceCache()
-    downloader = CacheDownloader("refs", disable_cache=disable_cache)
+    downloader = CacheDownloader(
+        "refs", disable_cache=disable_cache, url_rewrites=url_rewrites
+    )
 
     def get_local_file(uri: str) -> t.Any:
         path = filename2path(uri)

tests/acceptance/test_remote_ref_resolution.py

@@ -35,6 +35,41 @@
 }
 
 
+def test_remote_schema_and_refs_can_use_url_rewrites(run_line, tmp_path):
+    original_root = "https://schemas.example/"
+    mirror_root = "https://mirror.example/schemas/"
+    responses.add(
+        "GET",
+        f"{mirror_root}main.json",
+        json={
+            "$schema": "http://json-schema.org/draft-07/schema",
+            "properties": {"title": {"$ref": "./title.json"}},
+        },
+    )
+    responses.add("GET", f"{mirror_root}title.json", json={"type": "string"})
+    instance_path = tmp_path / "instance.json"
+    instance_path.write_text(json.dumps({"title": "rewritten"}))
+
+    result = run_line(
+        [
+            "check-jsonschema",
+            "--schemafile",
+            f"{original_root}main.json",
+            "--url-rewrite",
+            original_root,
+            mirror_root,
+            "--no-cache",
+            str(instance_path),
+        ]
+    )
+
+    assert result.exit_code == 0, result.output
+    assert [call.request.url for call in responses.calls] == [
+        f"{mirror_root}main.json",
+        f"{mirror_root}title.json",
+    ]
+
+
 @pytest.mark.parametrize("check_passes", (True, False))
 @pytest.mark.parametrize("casename", ("case1", "case2"))
 def test_remote_ref_resolution_simple_case(run_line, check_passes, casename, tmp_path):

tests/unit/cli/test_parse.py

@@ -109,6 +109,58 @@ def test_no_cache_flag_is_true(cli_runner, mock_parse_result, in_tmp_dir, tmp_pa
     assert mock_parse_result.disable_cache is True
 
 
+def test_url_rewrite_options_are_collected(
+    cli_runner, mock_parse_result, in_tmp_dir, tmp_path
+):
+    touch_files(tmp_path, "foo.json")
+    cli_runner.invoke(
+        cli_main,
+        [
+            "--schemafile",
+            "schema.json",
+            "--url-rewrite",
+            "https://schemas.example/",
+            "https://mirror.example/schemas/",
+            "--url-rewrite",
+            "https://schemas.example/special/",
+            "https://special.example/",
+            "foo.json",
+        ],
+    )
+
+    assert mock_parse_result.url_rewrites == (
+        ("https://schemas.example/", "https://mirror.example/schemas/"),
+        ("https://schemas.example/special/", "https://special.example/"),
+    )
+
+
+@pytest.mark.parametrize(
+    "source,target",
+    [
+        ("schemas.example/", "https://mirror.example/"),
+        ("https://schemas.example/", "/local/mirror/"),
+    ],
+)
+def test_url_rewrite_requires_absolute_http_urls(
+    cli_runner, source, target, in_tmp_dir, tmp_path
+):
+    touch_files(tmp_path, "foo.json")
+    result = cli_runner.invoke(
+        cli_main,
+        [
+            "--schemafile",
+            "schema.json",
+            "--url-rewrite",
+            source,
+            target,
+            "foo.json",
+        ],
+    )
+
+    assert result.exit_code == 2
+    assert "both prefixes must be absolute HTTP(S) URLs" in result.stderr
+
+
 @pytest.mark.parametrize(
     "cmd_args",
     [