@@ -6,29 +6,24 @@ use axum::Json;
 use crate::auth::AuthenticatedDid;
 use crate::error::{AppError, Result};
 use crate::state::AppState;
-use crate::visibility::{visibility_check, Decision};
 
 /// GET /api/v1/repos/{owner}/{repo}/encrypted-blobs
 /// Returns [{oid, cid}] for every encrypted blob in the repo, to any caller who
 /// can read the repo. Not recipient-scoped: recipient identities are not stored,
 /// so access control here is repo readability and decryption is gated by the
 /// envelope crypto (only a real recipient can open an envelope).
+///
+/// Quarantined repos are opaque 404 via [`crate::api::authorize_repo_read`] —
+/// same as issues/changelogs — so a public-but-quarantined mirror cannot leak
+/// its encrypted blob index.
 pub async fn list_encrypted_blobs(
     State(state): State<AppState>,
     auth: Option<Extension<AuthenticatedDid>>,
     Path((owner, repo)): Path<(String, String)>,
 ) -> Result<Json<serde_json::Value>> {
-    let record = state
-        .db
-        .get_repo(&owner, &repo)
-        .await?
-        .ok_or_else(|| AppError::RepoNotFound(format!("{owner}/{repo}")))?;
     let caller = auth.as_ref().map(|e| e.0 .0.as_str());
-    let rules = state.db.list_visibility_rules(&record.id).await?;
-    if visibility_check(&rules, record.is_public, &record.owner_did, caller, "/") == Decision::Deny
-    {
-        return Err(AppError::RepoNotFound(format!("{owner}/{repo}")));
-    }
+    let (record, _rules) =
+        crate::api::authorize_repo_read(&state, &owner, &repo, caller, "/").await?;
     let rows = state.db.list_all_encrypted_blobs(&record.id).await?;
     let blobs: Vec<_> = rows
         .into_iter()
@@ -45,17 +40,9 @@ pub async fn get_encrypted_blob(
     auth: Option<Extension<AuthenticatedDid>>,
     Path((owner, repo, oid)): Path<(String, String, String)>,
 ) -> Result<Vec<u8>> {
-    let record = state
-        .db
-        .get_repo(&owner, &repo)
-        .await?
-        .ok_or_else(|| AppError::RepoNotFound(format!("{owner}/{repo}")))?;
     let caller = auth.as_ref().map(|e| e.0 .0.as_str());
-    let rules = state.db.list_visibility_rules(&record.id).await?;
-    if visibility_check(&rules, record.is_public, &record.owner_did, caller, "/") == Decision::Deny
-    {
-        return Err(AppError::RepoNotFound(format!("{owner}/{repo}/{oid}")));
-    }
+    let (record, _rules) =
+        crate::api::authorize_repo_read(&state, &owner, &repo, caller, "/").await?;
     let cid = state
         .db
         .encrypted_blob_cid(&record.id, &oid)
@@ -81,17 +68,9 @@ pub async fn replicate_encrypted_blobs(
     auth: Option<Extension<AuthenticatedDid>>,
     Path((owner, repo)): Path<(String, String)>,
 ) -> Result<Json<serde_json::Value>> {
-    let record = state
-        .db
-        .get_repo(&owner, &repo)
-        .await?
-        .ok_or_else(|| AppError::RepoNotFound(format!("{owner}/{repo}")))?;
     let caller = auth.as_ref().map(|e| e.0 .0.as_str());
-    let rules = state.db.list_visibility_rules(&record.id).await?;
-    if visibility_check(&rules, record.is_public, &record.owner_did, caller, "/") == Decision::Deny
-    {
-        return Err(AppError::RepoNotFound(format!("{owner}/{repo}")));
-    }
+    let (record, _rules) =
+        crate::api::authorize_repo_read(&state, &owner, &repo, caller, "/").await?;
     let rows = state.db.list_all_encrypted_blobs(&record.id).await?;
     let blobs: Vec<_> = rows
         .into_iter()
 
@@ -40,16 +40,21 @@ use crate::visibility::{visibility_check, Decision};
 /// caller passes. For each iterated row we gate against that row's OWN rules
 /// (`visibility_check` at `"/"`), never re-resolving via `authorize_repo_read`
 /// — `get_repo`'s fuzzy match could otherwise authorize a different physical
-/// row than the one read (KTD2a). We check object existence via
-/// `store::object_type` *before* the expensive reachability walk so random-CID
-/// spray cannot trigger full-history git walks on repos that don't carry the
-/// object. When the row carries path-scoped rules (KTD4) the served object
-/// must be either a non-blob (trees/commits are structural; KTD3) OR a blob
-/// in the caller's *reachable* allowed-set (`allowed_blob_set_for_caller`).
-/// The reachable allowed-set excludes dangling blobs — a blob written via
-/// `git hash-object -w` and never committed has no path to gate, so it is
-/// fail-closed 404'd under path-scoped rules (#126). Denial and genuine
-/// not-found both fall through to an opaque 404.
+/// row than the one read (KTD2a). Quarantine is fail-closed in two layers:
+/// `list_all_repos` drops rows with `quarantined = TRUE`, and a logical-repo
+/// fold drops any surviving twin that shares owner+name with a quarantined
+/// *canonical* (UUID) row — so a public mirror cannot serve while its
+/// canonical twin is quarantined, without letting a quarantined mirror
+/// suppress a healthy canonical (`get_repo` prefers canonical). We check object
+/// existence via `store::object_type` *before* the expensive reachability walk
+/// so random-CID spray cannot trigger full-history git walks on repos that
+/// don't carry the object. When the row carries path-scoped rules (KTD4) the
+/// served object must be either a non-blob (trees/commits are structural;
+/// KTD3) OR a blob in the caller's *reachable* allowed-set
+/// (`allowed_blob_set_for_caller`). The reachable allowed-set excludes dangling
+/// blobs — a blob written via `git hash-object -w` and never committed has no
+/// path to gate, so it is fail-closed 404'd under path-scoped rules (#126).
+/// Denial and genuine not-found both fall through to an opaque 404.
 ///
 /// Scan completeness (F2): the 404 above is returned ONLY when every candidate
 /// repo reached a VERDICT — visibility deny, probe-says-absent, walk-gate deny,
@@ -86,6 +91,17 @@ use crate::visibility::{visibility_check, Decision};
 /// case. A stale-public mirror row still serves withheld content (tracked
 /// separately, #124).
 ///
+/// Normalize owner DID for quarantine logical-repo keys: matches [`did_matches`]
+/// for `did:key:` vs bare-key pairs without scanning every quarantined row per
+/// candidate repo.
+fn quarantine_owner_key(owner_did: &str) -> String {
+    match owner_did.strip_prefix("did:key:") {
+        Some(k) if !k.contains(':') => k.to_string(),
+        _ if !owner_did.contains(':') => owner_did.to_string(),
+        _ => owner_did.to_string(),
+    }
+}
+
 /// One `/ipfs` request's walk admission: the global pool permit plus the
 /// optional per-source sub-permit, both RAII (#174 U1).
 ///
@@ -216,6 +232,36 @@ pub async fn get_by_cid(
             )));
         }
     };
+    // Logical-repo quarantine fold: when a *canonical* UUID row is quarantined
+    // but a public mirror twin (`owner/name`) survives `list_all_repos`, slug
+    // routes 404 via `authorize_repo_read` (prefers canonical) while CID serve
+    // would otherwise still hit the mirror. Drop candidates that share
+    // owner+name with a quarantined *canonical* only — a quarantined mirror
+    // must not suppress a healthy canonical (same preference as get_repo).
+    let quarantined = match tokio::time::timeout(
+        request_deadline.saturating_duration_since(std::time::Instant::now()),
+        state.db.list_quarantined_repos(),
+    )
+    .await
+    {
+        Ok(Ok(rows)) => rows,
+        Ok(Err(e)) => return Err(AppError::Internal(e)),
+        Err(_elapsed) => {
+            tracing::warn!(
+                budget_secs = state.config.ipfs_request_budget_secs,
+                "/ipfs list_quarantined_repos exceeded the request budget \
+                 (GITLAWB_IPFS_REQUEST_BUDGET_SECS); shedding a retryable 503 and freeing the walk permit"
+            );
+            return Err(AppError::Overloaded(format!(
+                "ipfs scan incomplete (budget) for CID {cid_str}; retry shortly"
+            )));
+        }
+    };
+    let quarantined_canonical_keys: HashSet<(String, String)> = quarantined
+        .iter()
+        .filter(|q| !q.id.contains('/'))
+        .map(|q| (quarantine_owner_key(&q.owner_did), q.name.clone()))
+        .collect();
 
     // Fetch every repo's visibility rules in one query rather than one per row
     // (the gate runs each row against its OWN rules — KTD2a). A row absent from
@@ -320,6 +366,12 @@ pub async fn get_by_cid(
     let mut repos_visited: usize = 0;
 
     for repo in &repos {
+        if quarantined_canonical_keys
+            .contains(&(quarantine_owner_key(&repo.owner_did), repo.name.clone()))
+        {
+            continue;
+        }
+
         // Repo-level read gate against THIS row's own rules (KTD2a). Deny is a
         // VERDICT: this repo would never serve the caller, so skipping it cannot
         // hide content from them.
 
@@ -238,13 +238,16 @@ mod authz_guard {
             // PRE-GATED — already owner-gated, in-scope group; guard the gate itself
             (protect, "protect_branch", "did_matches("),
             (protect, "unprotect_branch", "did_matches("),
+            (visibility, "set_visibility", "authorize_repo_read("),
             (visibility, "set_visibility", "require_owner("),
+            (visibility, "remove_visibility", "authorize_repo_read("),
             (visibility, "remove_visibility", "require_owner("),
+            (visibility, "list_visibility", "authorize_repo_read("),
             (visibility, "list_visibility", "require_owner("),
         ];
 
-        // The visibility rows prove require_owner is CALLED; this proves the helper
-        // itself does DID-safe matching, not a raw/trailing-segment compare.
+        // Rows above prove each visibility handler calls authorize_repo_read and
+        // require_owner; this proves the helper itself does DID-safe matching.
         assert!(
             fn_body(visibility, "require_owner").contains("did_matches("),
             "visibility::require_owner must use did_matches for DID-safe owner matching"
 
@@ -85,11 +85,10 @@ pub async fn set_visibility(
     Path((owner, repo)): Path<(String, String)>,
     Json(req): Json<SetVisibilityRequest>,
 ) -> Result<(StatusCode, Json<serde_json::Value>)> {
-    let record = state
-        .db
-        .get_repo(&owner, &repo)
-        .await?
-        .ok_or_else(|| AppError::RepoNotFound(format!("{owner}/{repo}")))?;
+    // Quarantine first (via authorize_repo_read), then owner — same posture as
+    // list_visibility so a quarantined repo cannot be mutated while reads 404.
+    let (record, _rules) =
+        crate::api::authorize_repo_read(&state, &owner, &repo, Some(&auth.0), "/").await?;
     require_owner(&record, &auth.0)?;
     validate_path_glob(&req.path_glob)?;
 
@@ -141,11 +140,8 @@ pub async fn remove_visibility(
     Path((owner, repo)): Path<(String, String)>,
     Json(req): Json<RemoveVisibilityRequest>,
 ) -> Result<Json<serde_json::Value>> {
-    let record = state
-        .db
-        .get_repo(&owner, &repo)
-        .await?
-        .ok_or_else(|| AppError::RepoNotFound(format!("{owner}/{repo}")))?;
+    let (record, _rules) =
+        crate::api::authorize_repo_read(&state, &owner, &repo, Some(&auth.0), "/").await?;
     require_owner(&record, &auth.0)?;
 
     state
@@ -171,14 +167,12 @@ pub async fn list_visibility(
     Extension(auth): Extension<AuthenticatedDid>,
     Path((owner, repo)): Path<(String, String)>,
 ) -> Result<Json<serde_json::Value>> {
-    let record = state
-        .db
-        .get_repo(&owner, &repo)
-        .await?
-        .ok_or_else(|| AppError::RepoNotFound(format!("{owner}/{repo}")))?;
+    // Quarantine first (via authorize_repo_read), then owner — a quarantined
+    // mirror must be opaque even to a caller matching owner_did.
+    let (record, rules) =
+        crate::api::authorize_repo_read(&state, &owner, &repo, Some(&auth.0), "/").await?;
     require_owner(&record, &auth.0)?;
 
-    let rules = state.db.list_visibility_rules(&record.id).await?;
     let rules_json: Vec<_> = rules
         .into_iter()
         .map(|r| {
@@ -205,28 +199,18 @@ pub async fn list_visibility(
 /// denied one (`reinclude`), so a clean-clone client can sparse-exclude the
 /// denied subtrees while re-including the allowed nested paths. Unlike
 /// `list_visibility` this is not owner-gated and never exposes reader_dids.
+///
+/// Quarantined repos are opaque 404 via [`crate::api::authorize_repo_read`] —
+/// same posture as encrypted-blob discovery — so admission and private-subtree
+/// layout are not disclosed to anon, owner, or peers.
 pub async fn withheld_paths(
     State(state): State<AppState>,
     auth: Option<Extension<AuthenticatedDid>>,
     Path((owner, repo)): Path<(String, String)>,
 ) -> Result<Json<serde_json::Value>> {
-    let record = state
-        .db
-        .get_repo(&owner, &repo)
-        .await?
-        .ok_or_else(|| AppError::RepoNotFound(format!("{owner}/{repo}")))?;
-
-    let rules = state.db.list_visibility_rules(&record.id).await?;
     let caller = auth.as_ref().map(|e| e.0 .0.as_str());
-
-    // Whole-repo read gate: a caller who cannot read "/" gets repo-not-found,
-    // matching the git read endpoints, so this never discloses a private repo's
-    // existence or its path layout to an unauthorized caller.
-    if crate::visibility::visibility_check(&rules, record.is_public, &record.owner_did, caller, "/")
-        == crate::visibility::Decision::Deny
-    {
-        return Err(AppError::RepoNotFound(format!("{owner}/{repo}")));
-    }
+    let (record, rules) =
+        crate::api::authorize_repo_read(&state, &owner, &repo, caller, "/").await?;
 
     let withheld =
         crate::visibility::withheld_globs(&rules, record.is_public, &record.owner_did, caller);
 
@@ -1095,14 +1095,18 @@ impl Db {
 
     /// Raw list of every repo row — NOT deduped (a mirror row and its canonical
     /// row both appear) and without stars. For enumeration callers that must see
-    /// every physical row (e.g. the IPFS object scan in `api::ipfs`), not for
-    /// listing surfaces. Listing surfaces dedupe via `list_all_repos_deduped` or
-    /// `list_all_repos_with_stars` + `dedupe_canonical_repos`.
+    /// every *servable* physical row (e.g. the IPFS object scan in `api::ipfs`).
+    /// Quarantined mirrors are excluded — same invariant as
+    /// [`crate::api::authorize_repo_read`]: admitted but withheld from every
+    /// reader. Listing surfaces that also need star counts use
+    /// `list_all_repos_with_stars` / `list_all_repos_deduped`.
     pub async fn list_all_repos(&self) -> Result<Vec<RepoRecord>> {
         let rows = sqlx::query(
             "SELECT id, name, owner_did, description, is_public, default_branch,
                     created_at, updated_at, disk_path, forked_from, machine_id
-             FROM repos ORDER BY updated_at DESC",
+             FROM repos
+             WHERE quarantined = FALSE
+             ORDER BY updated_at DESC",
         )
         .fetch_all(&self.pool)
         .await?;
@@ -5329,6 +5333,34 @@ mod icaptcha_quarantine_tests {
         assert!(db.list_quarantined_repo_ids().await.unwrap().is_empty());
     }
 
+    /// IPFS/CID enumeration must not see quarantined rows (serve-side of the
+    /// authorize_repo_read quarantine short-circuit).
+    #[sqlx::test]
+    async fn list_all_repos_excludes_quarantined(pool: PgPool) {
+        let db = db(pool).await;
+        db.upsert_mirror_repo("z6owner", "live", "/srv/live", None, false)
+            .await
+            .unwrap();
+        db.upsert_mirror_repo("z6owner", "sick", "/srv/sick", None, true)
+            .await
+            .unwrap();
+        let ids: Vec<_> = db
+            .list_all_repos()
+            .await
+            .unwrap()
+            .into_iter()
+            .map(|r| r.id)
+            .collect();
+        assert!(
+            ids.iter().any(|id| id.contains("live")),
+            "live repo must be listed: {ids:?}"
+        );
+        assert!(
+            ids.iter().all(|id| !id.contains("sick")),
+            "quarantined repo must be excluded from list_all_repos: {ids:?}"
+        );
+    }
+
     /// A mirror admitted quarantined stays quarantined across a re-sync — the
     /// admission decision is made once and an operator's later release (or the
     /// initial quarantine) must not be reverted by ON CONFLICT.