diff --git a/README.md b/README.md
index e61321a..5268f37 100644
--- a/README.md
+++ b/README.md
@@ -142,6 +142,46 @@ Synthetic data is only schema-valid against the dictionary that generated it, so
 refuses a batch that doesn't match the version being uploaded (override with
 `--allow-version-mismatch`).
 
+## Verifying download access (check-download)
+
+Registration alone does not prove a file can be downloaded. Two failure modes
+are invisible until a user clicks the file in the portal: an Indexd record
+with no storage URL (nothing to download, ever), and a record Fence refuses
+to sign a URL for. `g3dt indexd check-download` walks the exact chain the
+portal hits — Indexd record → storage URL → DRS object → access methods →
+Fence signed URL — and reports PASS/FAIL per object, exiting non-zero if any
+object fails so it can gate a deployment step.
+
+Run it before a release, and after registering new files. The env selects the
+API key secret and the key's JWT selects the commons, so there is no URL to
+pass (and none to get wrong).
+
+```bash
+g3dt indexd check-download --env staging                    # sample the 25 newest
+g3dt indexd check-download --env staging --limit 50
+g3dt indexd check-download --env prod PREFIX/<uuid-1> PREFIX/<uuid-2>
+```
+
+With no GUIDs, the newest objects for the env's commons are sampled from the
+indexd registry (latest revision per baseid). The registry may live in a
+different AWS account than the commons being checked; if the env's AWS
+profile cannot reach it, pass GUIDs explicitly.
+
+Reading a failure:
+
+| Symptom | Meaning |
+|---|---|
+| `Indexd status: 404` | the object is not registered — a registration problem, not a download one |
+| `urls: []` / no access methods | registered but with no storage location; it can never download |
+| `Access endpoint … 401` | authorization: the API key's user lacks `read-storage` on the record's `authz` resource — an authz gap, not a broken key |
+| `Access endpoint … 500` | Fence has the permission but failed to sign — a service-side fault |
+
+On a 401, compare what the record requires
+(`https://commons.example.org/index/<did>`, the `authz` field) with what the
+key's user actually holds (`https://commons.example.org/user/user`):
+downloads require `read-storage` on the record's authz resource, which a user
+holding only `create` does not have.
+
 ## Development
 
 ```bash
diff --git a/src/g3dt/cli/indexd_cmds.py b/src/g3dt/cli/indexd_cmds.py
index c0f9d4d..51b3009 100644
--- a/src/g3dt/cli/indexd_cmds.py
+++ b/src/g3dt/cli/indexd_cmds.py
@@ -1,20 +1,26 @@
-"""`g3dt indexd` — register S3 files with Gen3 indexd (data-plane).
+"""`g3dt indexd` — register S3 files with Gen3 indexd, and verify they download.
 
-Long-running, so it supports ``--on ec2``.
+``register`` is a long data-plane op, so it supports ``--on ec2``.
+``check-download`` is a read-only HTTP check that takes seconds and whose
+whole value is the PASS/FAIL in your terminal, so it is local-only.
 """
 from __future__ import annotations
 
-from typing import List
+from typing import List, Optional
 
 import typer
 
-from g3dt.cli._internal import dispatch
+from g3dt.cli._internal import dispatch, runner
 from g3dt.cli._internal.dispatch import Target
-from g3dt.cli._internal.resolve import study_of
+from g3dt.cli._internal.resolve import env_of, study_of
 
-app = typer.Typer(no_args_is_help=True, help="Register files with Gen3 indexd.")
+app = typer.Typer(
+    no_args_is_help=True,
+    help="Register files with Gen3 indexd and verify download access.",
+)
 
 _REGISTER = "services/indexd/register_indexd.py"
+_CHECK_DOWNLOAD = "services/indexd/verify_file_access.py"
 
 
 @app.command()
@@ -67,3 +73,54 @@ def remote_cli(env_name):
     dispatch.run_or_dispatch(
         on, env, _REGISTER, build_args, "indexd-register", remote_cli=remote_cli,
     )
+
+
+@app.command(name="check-download")
+def check_download(
+    guids: Optional[List[str]] = typer.Argument(
+        None,
+        help="Object GUIDs, e.g. PREFIX/<uuid>. Omit to sample the most "
+             "recently registered objects from the indexd registry.",
+    ),
+    env: str = typer.Option(..., "--env", "-e", help="Environment, e.g. test."),
+    limit: int = typer.Option(
+        25, "--limit", "-n",
+        help="How many objects to sample when no GUIDs are given.",
+    ),
+    key_path: Optional[str] = typer.Option(
+        None, "--key-path",
+        help="Break-glass: local Gen3 API key JSON file, instead of the "
+             "env's secret.",
+    ),
+) -> None:
+    """Prove registered objects are downloadable end to end.
+
+    Walks Indexd -> DRS -> Fence signed URL for each GUID and exits non-zero
+    if any object fails, so it can gate a deployment step. Read-only and
+    local-only (seconds, not a long job — there is nothing to dispatch to EC2).
+
+    The env selects the API key secret and the key's JWT selects the commons,
+    so a staging env checks staging. There is no URL to pass.
+
+    With no GUIDs, the newest --limit objects for this commons are sampled
+    from the indexd registry (latest revision per baseid). The registry may
+    live in a different AWS account than the commons — sampling needs an env
+    whose AWS profile can reach it; otherwise pass GUIDs explicitly.
+
+    Examples:
+      g3dt indexd check-download --env staging                 # sample 25 newest
+      g3dt indexd check-download --env staging --limit 50
+      g3dt indexd check-download --env prod PREFIX/aaa PREFIX/bbb
+    """
+    # Validate the env before spawning a subprocess: an unknown env should
+    # fail here with the config error, not deep inside the script.
+    e = env_of(env)
+
+    args: List[str] = ["--env", e.name]
+    if key_path:
+        args += ["--key-path", key_path]
+    if guids:
+        args += list(guids)
+    else:
+        args += ["--limit", str(limit)]
+    runner.run(runner.python_script(_CHECK_DOWNLOAD, *args))
diff --git a/src/g3dt/indexd/file_access.py b/src/g3dt/indexd/file_access.py
new file mode 100644
index 0000000..941dece
--- /dev/null
+++ b/src/g3dt/indexd/file_access.py
@@ -0,0 +1,274 @@
+"""Verify that registered file objects are actually downloadable.
+
+Walks the full chain a user hits when they click a file in the portal:
+
+    Indexd record -> storage URLs -> DRS object -> access methods -> signed URL
+
+and reports PASS/FAIL per object. Two failure modes this catches that nothing
+else does:
+
+  - the Indexd record exists but has no URLs / DRS access methods, so the
+    object can never be downloaded;
+  - the object has a storage URL but Fence fails to sign a download URL
+    (e.g. HTTP 500 from the access endpoint).
+
+**The environment selects the credential, and the credential selects the
+commons.** The caller supplies an env; its ``aws_secret_name`` yields the Gen3
+API key (Secrets Manager, or a local file for the documented absolute-path
+convention), and the commons URL is derived from that key's JWT ``iss`` claim.
+There is therefore no URL for an operator to get wrong.
+
+That matters for a concrete reason: :class:`gen3.auth.Gen3Auth` silently falls
+back to the Workspace Token Service whenever an explicitly-passed ``endpoint``
+disagrees with a ``refresh_file``'s issuer (see ``gen3/auth.py``, the
+``elif refresh_file:`` branch). WTS is not deployed on these commons (e.g.
+``commons.example.org``), so that path fails with a misleading
+``502 Bad Gateway`` on ``/wts/external_oidc/``. Constructing
+``Gen3Auth(refresh_token=<key dict>)`` with no ``endpoint`` never enters that
+branch, which makes the mismatch structurally impossible rather than merely
+guarded.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any, Callable, List, Optional, Tuple
+
+import requests
+from gen3.auth import Gen3Auth
+
+from g3dt import config as g3dt_config
+from g3dt import resolver
+from g3dt.upload.metadata_submitter import (
+    commons_url_from_jwt,
+    create_boto3_session,
+    get_gen3_api_key_aws_secret,
+)
+
+
+def registry_sample_sql(
+    database: str, table: str, indexd_endpoint: str, limit: int
+) -> str:
+    """Build the SQL that samples the most recently registered objects.
+
+    Latest revision per ``baseid`` only (re-registering a file creates a new
+    ``did`` under the same ``baseid``; older revisions are superseded and
+    would fail a download check spuriously), filtered to one commons via
+    ``indexd_endpoint`` — that filter is what keeps another environment's
+    registrations out of the sample when environments share a registry table.
+    """
+    return f"""
+        SELECT did FROM (
+            SELECT did, registered_at,
+                ROW_NUMBER() OVER (
+                    PARTITION BY baseid
+                    ORDER BY registered_at DESC
+                ) AS row_num
+            FROM "{database}"."{table}"
+            WHERE indexd_endpoint = '{indexd_endpoint}'
+        )
+        WHERE row_num = 1
+        ORDER BY registered_at DESC
+        LIMIT {int(limit)}
+    """
+
+
+def sample_recent_guids(env_cfg, commons_url: str, limit: int) -> List[str]:
+    """Return the ``limit`` most recently registered GUIDs for a commons.
+
+    Queries the env's indexd registry table (Athena) with the env's AWS
+    profile. Every name comes from the resolver: the registry is the
+    conventional ``indexd_registry`` table in the env's metadata Glue DB, and
+    the Athena output location / workgroup are the env's own.
+
+    The registry may live in a different AWS account than the commons being
+    checked, so the env must be one whose ``aws_profile`` can reach it —
+    otherwise pass explicit GUIDs and skip sampling entirely.
+
+    The returned ``did`` values already include their prefix, so they feed
+    straight into :func:`verify_objects`.
+    """
+    from g3dt.utils.athena_utils import AthenaConfig, AthenaQuery
+
+    rc = resolver.resolve(
+        g3dt_config.require_project(),
+        g3dt_config.env_base(env_cfg.name),
+        profile=env_cfg.aws_profile,
+    )
+    database = rc.metadata_db
+    table = g3dt_config.INDEXD_REGISTRY_TABLE
+
+    sql = registry_sample_sql(
+        database, table, f"{commons_url}/index/index", limit
+    )
+    query = AthenaQuery(
+        AthenaConfig(
+            aws_region=env_cfg.region,
+            aws_profile=env_cfg.aws_profile,
+            athena_s3_output=rc.athena_output_location,
+            workgroup=rc.athena_workgroup,
+        )
+    )
+    try:
+        df = query.query_athena(sql, database, ctas_approach=False)
+    except Exception as exc:
+        raise RuntimeError(
+            f"could not query {database}.{table} with profile "
+            f"'{env_cfg.aws_profile}': {exc}. The indexd registry may live "
+            "in a different AWS account than the commons — use an env whose "
+            "AWS profile can reach the registry, or pass explicit GUIDs to "
+            "check-download instead."
+        ) from exc
+    return df["did"].dropna().tolist()
+
+
+def api_key_for_env(env_cfg, key_path: Optional[str] = None) -> dict:
+    """Load the Gen3 API key an environment authenticates with.
+
+    Precedence:
+      1. ``key_path`` — an explicit local key file (break-glass override);
+      2. the env's ``aws_secret_name`` as an absolute path (local file — the
+         documented path-style convention, as in ``register_indexd.py``);
+      3. the env's ``aws_secret_name`` as a Secrets Manager secret.
+
+    Args:
+        env_cfg: Resolved :class:`~g3dt.config.EnvConfig`.
+        key_path: Optional path to a local Gen3 API key JSON file.
+
+    Returns:
+        dict: The parsed API key, e.g. ``{"api_key": "<JWT>", ...}``.
+    """
+    if key_path:
+        with open(key_path, "r") as handle:
+            return json.load(handle)
+
+    secret_name = env_cfg.aws_secret_name
+    if secret_name.startswith("/"):
+        with open(secret_name, "r") as handle:
+            return json.load(handle)
+
+    session = create_boto3_session(
+        aws_profile=env_cfg.aws_profile,
+        aws_region=env_cfg.region,
+    )
+    return get_gen3_api_key_aws_secret(
+        secret_name=secret_name,
+        region_name=env_cfg.region,
+        session=session,
+    )
+
+
+def commons_auth(api_key: dict) -> Tuple[str, Gen3Auth]:
+    """Return the commons URL and an authenticated client for an API key.
+
+    Both are derived from the same token, so they cannot disagree. Note the
+    deliberate absence of an ``endpoint`` argument to ``Gen3Auth`` — passing
+    one alongside a credential is what triggers the Workspace Token Service
+    fallback described in the module docstring.
+
+    Returns:
+        tuple: ``(commons_url, Gen3Auth)``.
+    """
+    return commons_url_from_jwt(api_key["api_key"]), Gen3Auth(refresh_token=api_key)
+
+
+def get_json(
+    commons_url: str, auth: Gen3Auth, path: str
+) -> Tuple[requests.Response, Any]:
+    """GET a commons path, returning the response and its parsed body."""
+    response = requests.get(f"{commons_url}{path}", auth=auth, timeout=30)
+    try:
+        body = response.json()
+    except ValueError:
+        body = response.text
+    return response, body
+
+
+def verify_object(
+    commons_url: str,
+    auth: Gen3Auth,
+    object_id: str,
+    emit: Callable[[str], None] = print,
+) -> bool:
+    """Return True if one object is fully downloadable, else explain why not.
+
+    Args:
+        commons_url: Base URL of the commons to check.
+        auth: Authenticated Gen3 client.
+        object_id: The object GUID, e.g. ``PREFIX/<uuid>``.
+        emit: Where to write progress (defaults to ``print``).
+    """
+    emit(f"\n{'=' * 80}")
+    emit(f"Checking: {object_id}")
+
+    index_response, index_record = get_json(commons_url, auth, f"/index/{object_id}")
+    emit(f"Indexd status: {index_response.status_code}")
+    if not index_response.ok:
+        emit(str(index_record))
+        emit("FAIL: no Indexd record.")
+        return False
+
+    urls = index_record.get("urls", [])
+    emit(f"Indexd DID: {index_record.get('did')}")
+    emit(f"Indexd URLs: {json.dumps(urls, indent=2)}")
+
+    drs_response, drs_object = get_json(
+        commons_url, auth, f"/ga4gh/drs/v1/objects/{object_id}"
+    )
+    emit(f"DRS object status: {drs_response.status_code}")
+    if not drs_response.ok:
+        emit(str(drs_object))
+        emit("FAIL: no DRS object.")
+        return False
+
+    access_methods = drs_object.get("access_methods", [])
+    emit(f"DRS access methods: {json.dumps(access_methods, indent=2)}")
+
+    if not urls or not access_methods:
+        emit(
+            "FAIL: the Indexd record has no usable storage location, "
+            "so the object cannot be downloaded."
+        )
+        return False
+
+    ok = True
+    for method in access_methods:
+        access_id = method.get("access_id")
+        if not access_id:
+            continue
+
+        access_response, access_body = get_json(
+            commons_url, auth, f"/ga4gh/drs/v1/objects/{object_id}/access/{access_id}"
+        )
+        emit(f"Access endpoint ({access_id}) status: {access_response.status_code}")
+        emit(
+            "Access response: "
+            + (
+                json.dumps(access_body, indent=2)
+                if isinstance(access_body, (dict, list))
+                else str(access_body)
+            )
+        )
+        if not access_response.ok:
+            emit(
+                "FAIL: the object has a storage URL, but the server "
+                "failed to generate a usable download URL."
+            )
+            ok = False
+
+    if ok:
+        emit("PASS: object is downloadable.")
+    return ok
+
+
+def verify_objects(
+    commons_url: str,
+    auth: Gen3Auth,
+    object_ids: List[str],
+    emit: Callable[[str], None] = print,
+) -> List[str]:
+    """Verify several objects; return the GUIDs that failed (empty if all pass)."""
+    return [
+        object_id
+        for object_id in object_ids
+        if not verify_object(commons_url, auth, object_id, emit=emit)
+    ]
diff --git a/src/g3dt/services/indexd/verify_file_access.py b/src/g3dt/services/indexd/verify_file_access.py
new file mode 100644
index 0000000..8541f5a
--- /dev/null
+++ b/src/g3dt/services/indexd/verify_file_access.py
@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Verify that registered file objects are actually downloadable.
+
+Walks the chain a portal user hits — Indexd record -> storage URLs -> DRS
+object -> access methods -> Fence signed URL — and reports PASS/FAIL per
+object. Exits non-zero if any object fails, so it can gate a deployment step.
+
+The environment selects the credential and the credential selects the commons:
+``--env`` resolves the env's ``aws_secret_name`` (AWS Secrets Manager, or a
+local file when the value is an absolute path), and the commons URL comes from
+that key's JWT. There is no URL to pass and therefore none to get wrong — see
+``src/g3dt/indexd/file_access.py`` for why that matters.
+
+GUIDs are optional: with none given, the script samples the most recently
+registered objects for this commons from the indexd registry (Athena) — the
+latest revision per baseid, newest first, ``--limit`` of them. The registry
+may live in a different AWS account than the commons, so sampling needs an
+env whose AWS profile can reach it; explicit GUIDs skip Athena entirely.
+
+Prefer the CLI wrapper, which resolves the env the same way:
+
+    g3dt indexd check-download --env staging                # auto-sample
+    g3dt indexd check-download --env staging PREFIX/<uuid>
+
+Direct usage:
+
+    poetry run python src/g3dt/services/indexd/verify_file_access.py \
+        --env staging PREFIX/005d97ab-... PREFIX/ffff19f8-...
+"""
+
+import argparse
+import sys
+
+from g3dt import config as g3dt_config
+from g3dt.indexd.file_access import (
+    api_key_for_env,
+    commons_auth,
+    sample_recent_guids,
+    verify_objects,
+)
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument(
+        "guids",
+        nargs="*",
+        help="object GUIDs, e.g. PREFIX/uuid; omit to sample the registry",
+    )
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=25,
+        help="how many recently registered objects to sample when no GUIDs "
+        "are given (default: 25)",
+    )
+    parser.add_argument(
+        "--env",
+        required=True,
+        help="environment, e.g. staging (selects the API key secret)",
+    )
+    parser.add_argument(
+        "--key-path",
+        default=None,
+        help="break-glass: local Gen3 API key JSON file, instead of the env's secret",
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+
+    try:
+        env_cfg = g3dt_config.resolve_env(args.env)
+    except g3dt_config.ConfigError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 2
+
+    try:
+        api_key = api_key_for_env(env_cfg, args.key_path)
+    except Exception as exc:  # boto/OSError/JSON — all mean "no usable key"
+        print(
+            f"ERROR: could not load the Gen3 API key for env '{args.env}': {exc}",
+            file=sys.stderr,
+        )
+        return 2
+
+    commons, auth = commons_auth(api_key)
+    # Provenance: an operator can eyeball exactly what was checked, and with
+    # which credential, before trusting a PASS.
+    print(
+        f"Env: {env_cfg.name} | Secret: {args.key_path or env_cfg.aws_secret_name} "
+        f"| Commons: {commons}"
+    )
+
+    guids = args.guids
+    if not guids:
+        print(
+            f"No GUIDs given — sampling the {args.limit} most recently "
+            f"registered object(s) for {commons} from the indexd registry."
+        )
+        try:
+            guids = sample_recent_guids(env_cfg, commons, args.limit)
+        except RuntimeError as exc:
+            print(f"ERROR: {exc}", file=sys.stderr)
+            return 2
+        if not guids:
+            print(
+                f"ERROR: the registry has no objects for {commons}/index/index"
+                " — nothing to check. Register files first, or pass GUIDs.",
+                file=sys.stderr,
+            )
+            return 2
+        print(f"Sampled: {', '.join(guids)}")
+
+    failures = verify_objects(commons, auth, guids)
+
+    print(f"\n{'=' * 80}")
+    print(f"Checked {len(guids)} object(s): {len(failures)} failure(s).")
+    for guid in failures:
+        print(f"  FAILED: {guid}")
+    return 1 if failures else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/test_cli_commands.py b/tests/test_cli_commands.py
index 688a0be..96fced7 100644
--- a/tests/test_cli_commands.py
+++ b/tests/test_cli_commands.py
@@ -13,7 +13,7 @@
 from typer.testing import CliRunner
 
 from g3dt.cli.main import app
-from g3dt.config import EnvConfig, StudyConfig, dictionary_url
+from g3dt.config import ConfigError, EnvConfig, StudyConfig, dictionary_url
 
 runner = CliRunner()
 
@@ -378,3 +378,74 @@ def test_k8s_restart_schema_passes_env_argo_args(mock_run, _env):
     assert argv[0] == "bash"
     assert argv[1].endswith("services/k8s_ops/argocd_restart_schema.sh")
     assert "-d" in argv and "-a" in argv and "-n" in argv
+
+
+@patch("g3dt.config.resolve_env", side_effect=ConfigError("No SSM parameters found under /etl/nope"))
+@patch("g3dt.cli._internal.runner.run")
+def test_indexd_check_download_validates_env_before_spawning(mock_run, _resolve):
+    """
+    Background:
+        check-download resolves the env locally (env_of) before it spawns the
+        worker script, so an unknown env fails right here with the config
+        error — not seconds later, deep inside a subprocess whose traceback
+        buries the real problem.
+
+    Inputs:  g3dt indexd check-download --env nope, where env resolution
+             raises ConfigError
+    Expected Output: a clean non-zero exit carrying the config message; the
+    worker subprocess is never invoked.
+    """
+    result = runner.invoke(app, ["indexd", "check-download", "--env", "nope"])
+    assert result.exit_code == 1
+    mock_run.assert_not_called()
+    assert "No SSM parameters found" in result.output
+
+
+@patch("g3dt.cli.indexd_cmds.env_of", side_effect=_env_cfg)
+@patch("g3dt.cli._internal.runner.run")
+def test_indexd_check_download_forwards_guids_in_order(mock_run, _env):
+    """
+    Background:
+        Explicit GUIDs must bypass registry sampling entirely (they need no
+        Athena access — e.g. spot-checking an object that just failed in the
+        portal), so --limit is NOT forwarded alongside them.
+
+    Inputs:  g3dt indexd check-download --env staging PREFIX/aaa PREFIX/bbb
+    Expected Output:
+      - exit code 0
+      - runs verify_file_access.py via the current interpreter with the env
+        and both GUIDs as trailing positionals, in the order given, and no
+        --limit flag
+    """
+    result = runner.invoke(
+        app,
+        ["indexd", "check-download", "--env", "staging",
+         "PREFIX/aaa", "PREFIX/bbb"],
+    )
+    assert result.exit_code == 0, result.output
+    argv = _argv(mock_run)
+    assert argv[0] == sys.executable
+    assert argv[1].endswith("services/indexd/verify_file_access.py")
+    assert argv[2:] == ["--env", "staging", "PREFIX/aaa", "PREFIX/bbb"]
+    assert "--limit" not in argv
+
+
+@patch("g3dt.cli.indexd_cmds.env_of", side_effect=_env_cfg)
+@patch("g3dt.cli._internal.runner.run")
+def test_indexd_check_download_samples_registry_with_limit(mock_run, _env):
+    """
+    Background:
+        With no GUIDs given, the wrapped script samples the newest registered
+        objects for the env's commons from the indexd registry, so the
+        default pre-flight is a single command with no copy-paste. The CLI's
+        --limit controls the sample size (default 25).
+
+    Inputs:  g3dt indexd check-download --env staging --limit 5   (no GUIDs)
+    Expected Output: --limit 5 forwarded and NO trailing GUID positionals,
+    which tells the script to sample the registry itself.
+    """
+    result = runner.invoke(
+        app, ["indexd", "check-download", "--env", "staging", "--limit", "5"]
+    )
+    assert result.exit_code == 0, result.output
+    assert _argv(mock_run)[2:] == ["--env", "staging", "--limit", "5"]
diff --git a/tests/test_file_access.py b/tests/test_file_access.py
new file mode 100644
index 0000000..8e786fa
--- /dev/null
+++ b/tests/test_file_access.py
@@ -0,0 +1,440 @@
+"""Tests for the indexd file-access verification logic.
+
+These cover how the Gen3 credential and the commons URL are resolved (the part
+that caused a real incident — see test_commons_auth_never_passes_endpoint), and
+the four outcomes of walking the download chain. All AWS and HTTP calls are
+mocked; nothing touches a real commons.
+"""
+import base64
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from g3dt.config import EnvConfig
+from g3dt.indexd.file_access import (
+    api_key_for_env,
+    commons_auth,
+    verify_object,
+    verify_objects,
+)
+from g3dt.resolver import ResolvedConfig
+
+MODULE = "g3dt.indexd.file_access"
+
+
+def _b64(obj):
+    return base64.urlsafe_b64encode(json.dumps(obj).encode()).decode().rstrip("=")
+
+
+def _fake_jwt(iss):
+    """Build a structurally valid (unsigned) JWT carrying an `iss` claim."""
+    header = _b64({"alg": "RS256", "typ": "JWT"})
+    sig = base64.urlsafe_b64encode(b"sig").decode().rstrip("=")
+    return f"{header}.{_b64({'iss': iss})}.{sig}"
+
+
+STAGING_KEY = {"api_key": _fake_jwt("https://staging.commons.example.org/user")}
+
+
+def _env_cfg(secret_name="gen3_api_key_staging"):
+    return EnvConfig(
+        name="staging",
+        is_ec2=False,
+        region="ap-southeast-2",
+        dictionary_version="v1",
+        aws_profile="etl_staging",
+        aws_secret_name=secret_name,
+        schema_s3_uri="u",
+        domain="d",
+        app_name="a",
+        namespace="n",
+        cluster_name="c",
+        schema_repo="Org/schema-repo",
+    )
+
+
+def _response(status, body):
+    """A stand-in for a requests.Response with a JSON body."""
+    resp = MagicMock()
+    resp.status_code = status
+    resp.ok = 200 <= status < 300
+    resp.json.return_value = body
+    return resp
+
+
+# ---------------------------------------------------------------------------
+# Credential resolution
+# ---------------------------------------------------------------------------
+
+@patch(f"{MODULE}.get_gen3_api_key_aws_secret")
+@patch(f"{MODULE}.create_boto3_session")
+def test_api_key_for_env_fetches_the_envs_secret(mock_session, mock_load):
+    """
+    Test that the environment's configured secret is what gets loaded.
+
+    Background:
+        The CLI passes only an env; everything else must be derived from the
+        env's config so a staging run can never authenticate with a prod key.
+
+    Inputs:  an EnvConfig whose aws_secret_name is 'gen3_api_key_staging'
+    Expected Output:
+      - a boto3 session built with the env's profile and region
+      - get_gen3_api_key_aws_secret called with that secret name and region
+      - the returned key dict passed straight back to the caller
+    """
+    mock_load.return_value = STAGING_KEY
+
+    result = api_key_for_env(_env_cfg())
+
+    assert result is STAGING_KEY
+    mock_session.assert_called_once_with(
+        aws_profile="etl_staging", aws_region="ap-southeast-2"
+    )
+    assert mock_load.call_args.kwargs["secret_name"] == "gen3_api_key_staging"
+    assert mock_load.call_args.kwargs["region_name"] == "ap-southeast-2"
+
+
+@patch(f"{MODULE}.get_gen3_api_key_aws_secret")
+@patch(f"{MODULE}.create_boto3_session")
+def test_api_key_for_env_prefers_an_explicit_key_file(mock_session, mock_load, tmp_path):
+    """
+    Test the break-glass local key override.
+
+    Background:
+        An operator debugging a broken secret needs a way to test with a key
+        downloaded from the portal. When that path is given, no AWS call should
+        happen at all.
+
+    Inputs:  key_path pointing at a local JSON key file
+    Expected Output: the file's contents are returned; Secrets Manager and the
+    boto3 session are never touched.
+    """
+    key_file = tmp_path / "key.json"
+    key_file.write_text(json.dumps(STAGING_KEY))
+
+    result = api_key_for_env(_env_cfg(), key_path=str(key_file))
+
+    assert result == STAGING_KEY
+    mock_session.assert_not_called()
+    mock_load.assert_not_called()
+
+
+@patch(f"{MODULE}.get_gen3_api_key_aws_secret")
+@patch(f"{MODULE}.create_boto3_session")
+def test_api_key_for_env_reads_path_style_secret_name_as_local_file(
+    mock_session, mock_load, tmp_path
+):
+    """
+    Test the documented absolute-path convention for aws_secret_name.
+
+    Background:
+        An env may declare its aws_secret_name as an absolute path — the
+        dual-mode convention register_indexd.py already honours: a path-style
+        value is a local Gen3 API key file, anything else is a Secrets
+        Manager secret name. check-download must follow the same rule so an
+        env that registers files can always verify them.
+
+    Inputs:  an EnvConfig whose aws_secret_name is '/abs/path/key.json'
+    Expected Output: the file's contents are returned; no AWS call happens.
+    """
+    key_file = tmp_path / "key.json"
+    key_file.write_text(json.dumps(STAGING_KEY))
+
+    result = api_key_for_env(_env_cfg(secret_name=str(key_file)))
+
+    assert result == STAGING_KEY
+    mock_session.assert_not_called()
+    mock_load.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Commons resolution — the incident this feature exists to prevent
+# ---------------------------------------------------------------------------
+
+@patch(f"{MODULE}.Gen3Auth")
+def test_commons_auth_derives_url_from_the_key(mock_gen3auth):
+    """
+    Test that the commons URL comes from the API key's JWT issuer.
+
+    Background:
+        The key IS the environment selector. A staging key must produce the
+        staging commons without anyone typing a URL.
+
+    Inputs:  an API key whose iss is 'https://staging.commons.example.org/user'
+    Expected Output: commons == 'https://staging.commons.example.org'
+    """
+    commons, _auth = commons_auth(STAGING_KEY)
+    assert commons == "https://staging.commons.example.org"
+
+
+@patch(f"{MODULE}.Gen3Auth")
+def test_commons_auth_never_passes_endpoint(mock_gen3auth):
+    """
+    Test that Gen3Auth is constructed with the key alone — no endpoint.
+
+    Background:
+        This is the whole reason the command resolves credentials this way.
+        gen3/auth.py switches to the Workspace Token Service whenever an
+        explicit `endpoint` disagrees with a credential's issuer. WTS is not
+        deployed on these commons, so that path fails with a misleading
+        '502 Bad Gateway' on /wts/external_oidc/ — which reads like broken
+        object_id links but is purely a client-side mismatch. Passing
+        refresh_token and NO endpoint never enters that branch, making the
+        mismatch structurally impossible instead of merely guarded.
+
+    Inputs:  a staging API key
+    Expected Output: Gen3Auth called once with refresh_token=<key> only; no
+    'endpoint' and no 'refresh_file' keyword anywhere in the call.
+    """
+    commons_auth(STAGING_KEY)
+
+    mock_gen3auth.assert_called_once_with(refresh_token=STAGING_KEY)
+    assert "endpoint" not in mock_gen3auth.call_args.kwargs
+    assert "refresh_file" not in mock_gen3auth.call_args.kwargs
+
+
+# ---------------------------------------------------------------------------
+# The download chain
+# ---------------------------------------------------------------------------
+
+@patch(f"{MODULE}.requests.get")
+def test_verify_object_passes_when_the_whole_chain_resolves(mock_get):
+    """
+    Test the happy path: Indexd -> DRS -> signed URL all succeed.
+
+    Inputs:  an Indexd record with a storage URL, a DRS object with one access
+             method, and an access endpoint that returns a signed URL
+    Expected Output: True, and the final line emitted is the PASS message.
+    """
+    mock_get.side_effect = [
+        _response(200, {"did": "PREFIX/aaa", "urls": ["s3://bucket/f.csv"]}),
+        _response(200, {"access_methods": [{"access_id": "s3"}]}),
+        _response(200, {"url": "https://signed"}),
+    ]
+    lines = []
+
+    assert verify_object("https://commons", MagicMock(), "PREFIX/aaa", emit=lines.append)
+    assert lines[-1] == "PASS: object is downloadable."
+
+
+@patch(f"{MODULE}.requests.get")
+def test_verify_object_fails_when_there_is_no_indexd_record(mock_get):
+    """
+    Test the unregistered-object case.
+
+    Inputs:  Indexd returns 404
+    Expected Output: False, and no further calls are made (no point asking DRS
+    about an object that does not exist).
+    """
+    mock_get.side_effect = [_response(404, {"error": "no record"})]
+    lines = []
+
+    assert not verify_object("https://commons", MagicMock(), "PREFIX/aaa", emit=lines.append)
+    assert mock_get.call_count == 1
+    assert "FAIL: no Indexd record." in lines
+
+
+@patch(f"{MODULE}.requests.get")
+def test_verify_object_fails_when_the_record_has_no_storage_location(mock_get):
+    """
+    Test the 'registered but undownloadable' failure mode.
+
+    Background:
+        An Indexd record can exist with an empty `urls` list — the metadata
+        looks fine in the portal, but there is nothing to download. This is one
+        of the two real failure modes this checker exists to catch.
+
+    Inputs:  Indexd 200 with urls=[]; DRS 200 with access_methods=[]
+    Expected Output: False, with the 'no usable storage location' explanation.
+    """
+    mock_get.side_effect = [
+        _response(200, {"did": "PREFIX/aaa", "urls": []}),
+        _response(200, {"access_methods": []}),
+    ]
+    lines = []
+
+    assert not verify_object("https://commons", MagicMock(), "PREFIX/aaa", emit=lines.append)
+    assert any("no usable storage location" in line for line in lines)
+
+
+@patch(f"{MODULE}.requests.get")
+def test_verify_object_fails_when_fence_cannot_sign(mock_get):
+    """
+    Test the 'storage URL exists but Fence errors' failure mode.
+
+    Background:
+        The second real failure mode: the record and DRS object are healthy,
+        but the access endpoint 500s, so a user still cannot download the file.
+
+    Inputs:  a healthy record and DRS object, access endpoint returns 500
+    Expected Output: False, with the 'failed to generate a usable download URL'
+    explanation.
+    """
+    mock_get.side_effect = [
+        _response(200, {"did": "PREFIX/aaa", "urls": ["s3://bucket/f.csv"]}),
+        _response(200, {"access_methods": [{"access_id": "s3"}]}),
+        _response(500, {"error": "service failure"}),
+    ]
+    lines = []
+
+    assert not verify_object("https://commons", MagicMock(), "PREFIX/aaa", emit=lines.append)
+    assert any("failed to generate a usable download URL" in line for line in lines)
+
+
+@patch(f"{MODULE}.verify_object")
+def test_verify_objects_returns_only_the_failed_guids(mock_verify):
+    """
+    Test that the caller can derive an exit code from the result.
+
+    Background:
+        The script exits non-zero when any object fails, so this command can
+        gate a deployment step. That depends on failures being reported
+        precisely rather than as a bare boolean.
+
+    Inputs:  three GUIDs where the middle one fails
+    Expected Output: ['PREFIX/bbb'] — only the failure, order preserved.
+    """
+    mock_verify.side_effect = [True, False, True]
+
+    failures = verify_objects(
+        "https://commons", MagicMock(), ["PREFIX/aaa", "PREFIX/bbb", "PREFIX/ccc"]
+    )
+
+    assert failures == ["PREFIX/bbb"]
+
+
+# ---------------------------------------------------------------------------
+# Registry sampling (auto-extracted GUIDs for check-download)
+# ---------------------------------------------------------------------------
+
+def _rc():
+    """The slice of the env's SSM tree that sampling reads (resolver names)."""
+    return ResolvedConfig(
+        project="etl",
+        env="staging",
+        params={
+            "glue/db/metadata": "etl_staging_metadata_db",
+            "athena/outputLocation": "s3://etl-staging-metadata/athena-query-results/",
+            "athena/workgroup": "etl_staging_wg",
+        },
+    )
+
+
+def test_registry_sample_sql_scopes_to_one_commons_and_latest_revision():
+    """
+    Test that the sampling SQL cannot leak objects from another environment
+    or return superseded revisions.
+
+    Background:
+        Environments can write to a shared registry table, distinguished
+        only by the indexd_endpoint column — without that filter a "prod"
+        check would happily sample staging objects and prove nothing. And
+        re-registering a file creates a new did under the same baseid, so
+        without the latest-revision window function the sample could contain
+        old dids that legitimately no longer download.
+
+    Inputs:  database, table, a prod indexd endpoint, limit 10
+    Expected Output: SQL containing the endpoint equality filter, the
+    ROW_NUMBER window partitioned by baseid, and LIMIT 10.
+    """
+    from g3dt.indexd.file_access import registry_sample_sql
+
+    sql = registry_sample_sql(
+        "etl_staging_metadata_db",
+        "indexd_registry",
+        "https://commons.example.org/index/index",
+        10,
+    )
+
+    assert "indexd_endpoint = 'https://commons.example.org/index/index'" in sql
+    assert "PARTITION BY baseid" in sql
+    assert "row_num = 1" in sql
+    assert "LIMIT 10" in sql
+
+
+@patch("g3dt.utils.athena_utils.AthenaQuery")
+@patch(f"{MODULE}.resolver.resolve")
+@patch(f"{MODULE}.g3dt_config.require_project", return_value="etl")
+def test_sample_recent_guids_returns_dids_for_the_derived_endpoint(
+    _project, mock_resolve, mock_query_cls
+):
+    """
+    Test that sampling queries the registry with the env's own AWS session
+    and resolver-provided names, and returns plain did strings ready for
+    verify_objects.
+
+    Background:
+        The dids in the registry already include the "PREFIX/" prefix, so
+        the sample must be forwarded untouched — historically operators
+        double-prefixed GUIDs pasted from Athena and got spurious 404s. The
+        registry database, Athena output location and workgroup all come
+        from SSM via the resolver — nothing is hard-coded.
+
+    Inputs:  a resolver returning the env's metadata DB and Athena names, a
+             staging env, commons https://staging.commons.example.org, limit 2
+    Expected Output: the two dids from the query result, in order; the SQL
+    passed to Athena filters on the commons URL + /index/index; the resolver
+    is asked for this project/env with the env's profile.
+    """
+    import pandas as pd
+
+    from g3dt.indexd.file_access import sample_recent_guids
+
+    mock_resolve.return_value = _rc()
+    mock_query_cls.return_value.query_athena.return_value = pd.DataFrame(
+        {"did": ["PREFIX/aaa", "PREFIX/bbb"]}
+    )
+
+    guids = sample_recent_guids(
+        _env_cfg(), "https://staging.commons.example.org", 2
+    )
+
+    assert guids == ["PREFIX/aaa", "PREFIX/bbb"]
+    sql = mock_query_cls.return_value.query_athena.call_args[0][0]
+    assert "https://staging.commons.example.org/index/index" in sql
+    assert '"etl_staging_metadata_db"."indexd_registry"' in sql
+    mock_resolve.assert_called_once_with("etl", "staging", profile="etl_staging")
+    athena_cfg = mock_query_cls.call_args[0][0]
+    assert athena_cfg.athena_s3_output == (
+        "s3://etl-staging-metadata/athena-query-results/"
+    )
+    assert athena_cfg.workgroup == "etl_staging_wg"
+
+
+@patch("g3dt.utils.athena_utils.AthenaQuery")
+@patch(f"{MODULE}.resolver.resolve")
+@patch(f"{MODULE}.g3dt_config.require_project", return_value="etl")
+def test_sample_recent_guids_explains_cross_account_athena_failures(
+    _project, mock_resolve, mock_query_cls
+):
+    """
+    Test that an Athena failure surfaces as advice, not a raw stack trace.
+
+    Background:
+        The registry table may live in a different AWS account than the
+        commons being checked (e.g. a prod commons whose registry sits in a
+        shared staging account). Running check-download with a profile that
+        authenticates against the wrong account is the most likely sampling
+        failure in practice. The error must tell the operator the two ways
+        out: use an env whose profile can reach the registry, or pass
+        explicit GUIDs and skip sampling.
+
+    Inputs:  AthenaQuery.query_athena raising an exception
+    Expected Output: RuntimeError explaining the cross-account possibility
+    and naming explicit GUIDs as the fallback.
+    """
+    from g3dt.indexd.file_access import sample_recent_guids
+
+    mock_resolve.return_value = _rc()
+    mock_query_cls.return_value.query_athena.side_effect = Exception(
+        "EntityNotFoundException: Database etl_staging_metadata_db not found"
+    )
+
+    with pytest.raises(RuntimeError) as excinfo:
+        sample_recent_guids(
+            _env_cfg(), "https://staging.commons.example.org", 10
+        )
+
+    assert "different AWS account" in str(excinfo.value)
+    assert "explicit GUIDs" in str(excinfo.value)