Remove AR_ACCOUNT_L1=legacy, a second key to L1 that nobody tracked

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## [Unreleased]
+
+### Removed
+- **`AR_ACCOUNT_L1=legacy`**: an environment variable that turned any login into an L1
+  grant on `/fetch-field`, `/fetch-field-wkt`, `/fetch-field-centroid` and
+  `/geoid/{id}/eudr-export`. It was a development convenience and would have been a
+  production bypass had it ever been set there; it was documented nowhere. The variable
+  is no longer read. Two tests set it and require the L0 answer on all four routes; a
+  third shows a real grant still unlocks L1 under the same conditions.
+
 ## [v0.9-review]
 
 ### Added

app/routers/fetch_field.py

@@ -161,10 +161,7 @@ async def fetch_field(
                 detail="Field not found, invalid Geo Id."
             )
 
-        import os
         has_l1 = bool(x_field_grant and verify_field_grant(x_field_grant, record.geo_id))
-        if not has_l1 and os.getenv("AR_ACCOUNT_L1") == "legacy" and user is not None:
-            has_l1 = True
 
         return apply_masking_logic(record, {"granted": True} if has_l1 else None, s2_index)
 
@@ -187,10 +184,7 @@ async def fetch_field_wkt(geo_id: str, user: dict | None = Depends(get_current_u
         if not record:
             raise HTTPException(status_code=404, detail="Field not found.")
 
-        import os
         has_l1 = bool(x_field_grant and verify_field_grant(x_field_grant, record.geo_id))
-        if not has_l1 and os.getenv("AR_ACCOUNT_L1") == "legacy" and user is not None:
-            has_l1 = True
 
         if has_l1:
             wkt = record.geo_data.get('wkt')
@@ -228,10 +222,7 @@ async def fetch_field_centroid(geo_id: str, user: dict | None = Depends(get_curr
 
         centroid = Utils.fetch_field_centroid_by_wkt(field_wkt)
 
-        import os
         has_l1 = bool(x_field_grant and verify_field_grant(x_field_grant, record.geo_id))
-        if not has_l1 and os.getenv("AR_ACCOUNT_L1") == "legacy" and user is not None:
-            has_l1 = True
 
         if has_l1:
             return {
@@ -268,10 +259,7 @@ async def eudr_export(geo_id: str, user: dict | None = Depends(get_current_user)
         if not record:
             raise HTTPException(status_code=404, detail="Field not found.")
 
-        import os
         has_l1 = bool(x_field_grant and verify_field_grant(x_field_grant, record.geo_id))
-        if not has_l1 and os.getenv("AR_ACCOUNT_L1") == "legacy" and user is not None:
-            has_l1 = True
 
         if not has_l1:
             raise HTTPException(

app/tests/test_api.py

@@ -87,12 +87,14 @@ def test_register_and_fetch_field(mock_auth):
     assert res_l1.status_code == 200
     assert res_l1.json()["MaskingLevel"] == "L0"
 
-    print("  -> Step 5: Fetching with Valid Token in AR_ACCOUNT_L1=legacy mode (L1)")
+    print("  -> Step 5: AR_ACCOUNT_L1=legacy no longer unlocks L1 for a login without a grant")
     os.environ["AR_ACCOUNT_L1"] = "legacy"
-    res_l1_leg = client.get(f"/resolve/{geo_id}", headers={"Authorization": "Bearer valid-token"})
+    try:
+        res_l1_leg = client.get(f"/resolve/{geo_id}", headers={"Authorization": "Bearer valid-token"})
+    finally:
+        os.environ.pop("AR_ACCOUNT_L1", None)
     assert res_l1_leg.status_code == 200
-    assert res_l1_leg.json()["MaskingLevel"] == "L1"
-    os.environ.pop("AR_ACCOUNT_L1", None)
+    assert res_l1_leg.json()["MaskingLevel"] == "L0"
     print("[TEST] test_register_and_fetch_field SUCCESS")
 
 def test_fetch_field_wkt(mock_auth):
@@ -274,10 +276,12 @@ def get_signing_key_from_jwt(self, token):
         assert res1.json()["MaskingLevel"] == "L0"
 
         os.environ["AR_ACCOUNT_L1"] = "legacy"
-        res1_leg = client.get(f"/resolve/{geo_id}", headers={"Authorization": f"Bearer {token1}"})
+        try:
+            res1_leg = client.get(f"/resolve/{geo_id}", headers={"Authorization": f"Bearer {token1}"})
+        finally:
+            os.environ.pop("AR_ACCOUNT_L1", None)
         assert res1_leg.status_code == 200
-        assert res1_leg.json()["MaskingLevel"] == "L1"
-        os.environ.pop("AR_ACCOUNT_L1", None)
+        assert res1_leg.json()["MaskingLevel"] == "L0", "a real RS256 login plus the legacy flag is still not a grant"
 
         res2 = client.get(f"/resolve/{geo_id}", headers={"Authorization": f"Bearer {token2}"})
         assert res2.status_code == 200
@@ -429,3 +433,52 @@ def test_identity_resolution_nested(mock_auth):
     alias = db.query(GeoIDAlias).filter(GeoIDAlias.canonical_geo_id == parent_geo_id, GeoIDAlias.relation == "child_of").first()
     assert alias is not None
     db.close()
+
+
+# ---------------------------------------------------------------------------
+# AR_ACCOUNT_L1=legacy. Until 2026-10 this environment variable turned every
+# login into an L1 grant on the four field routes: boundary, WKT, centroid and
+# the EU filing. It was a development convenience that would have been a
+# production bypass had it ever been set. The variable is no longer read
+# anywhere. These tests set it, present a valid login and no grant, and
+# require the L0 answer on every route -- and then show that a real grant on
+# the same field still unlocks L1, so the L0 is a refusal and not an outage.
+# ---------------------------------------------------------------------------
+
+def test_the_legacy_flag_unlocks_nothing_on_any_field_route(mock_auth, monkeypatch):
+    monkeypatch.setenv("AR_ACCOUNT_L1", "legacy")
+    res = client.post("/register-field-boundary",
+                      json={"wkt": TEST_POLYGON_WKT, "threshold": 95, "return_s2_indices": False})
+    geo_id = extract_geo_id(res)
+    login = {"Authorization": "Bearer valid-token"}
+
+    r = client.get(f"/resolve/{geo_id}", headers=login)
+    assert r.status_code == 200 and r.json()["MaskingLevel"] == "L0", r.text
+
+    r = client.get(f"/fetch-field-wkt/{geo_id}", headers=login)
+    assert r.status_code == 200 and r.json()["WKT"] is None, r.text
+
+    r = client.get(f"/fetch-field-centroid/{geo_id}", headers=login)
+    assert r.status_code == 200 and r.json()["MaskingLevel"] == "L0", r.text
+
+    r = client.get(f"/geoid/{geo_id}/eudr-export", headers=login)
+    assert r.status_code == 401, r.text
+
+
+def test_a_grant_still_unlocks_l1_so_the_refusal_above_is_a_refusal(mock_auth, monkeypatch, setup_database):
+    """The check can fire: same routes, same login, plus a grant for the field."""
+    monkeypatch.setenv("AR_ACCOUNT_L1", "legacy")
+    geoid = "3f1a9f0f36e44c0cb1ad4c2f8e3a7d6b1c5e9d8f7a6b5c4d3e2f1a0b9c8d7e6f"  # the field valid.sdjwt names
+    grant = {"X-Field-Grant": get_token("valid.sdjwt"), "Authorization": "Bearer valid-token"}
+
+    r = client.get(f"/resolve/{geoid}", headers=grant)
+    assert r.status_code == 200 and r.json()["MaskingLevel"] == "L1", r.text
+
+    r = client.get(f"/fetch-field-wkt/{geoid}", headers=grant)
+    assert r.status_code == 200 and r.json()["WKT"], r.text
+
+    r = client.get(f"/fetch-field-centroid/{geoid}", headers=grant)
+    assert r.status_code == 200 and r.json()["MaskingLevel"] == "L1", r.text
+
+    r = client.get(f"/geoid/{geoid}/eudr-export", headers=grant)
+    assert r.status_code == 200, r.text