fix(validator): reject structurally invalid PAT stores

gittensor/validator/pat_handler.py

@@ -91,7 +91,7 @@ def _reject(reason: str) -> PatBroadcastSynapse:
 
         # 5. Store PAT (github_id guaranteed non-None after validate_github_credentials success)
         pat_storage.save_pat(uid=uid, hotkey=hotkey, pat=synapse.github_access_token, github_id=github_id or '0')
-    except (json.JSONDecodeError, OSError) as e:
+    except (json.JSONDecodeError, OSError, pat_storage.PatStoreFormatError) as e:
         return _reject(f'Validator PAT store temporarily unavailable; please retry. ({e})')
 
     # Clear PAT from response so it isn't echoed back
@@ -136,7 +136,7 @@ async def handle_pat_check(validator: 'Validator', synapse: PatCheckSynapse) ->
     # mislead the miner into re-broadcasting against an unreadable store).
     try:
         entry = pat_storage.get_pat_by_uid(uid)
-    except (json.JSONDecodeError, OSError) as e:
+    except (json.JSONDecodeError, OSError, pat_storage.PatStoreFormatError) as e:
         synapse.has_pat = False
         synapse.pat_valid = None
         synapse.rejection_reason = 'Validator PAT store temporarily unavailable; please retry.'

gittensor/validator/pat_storage.py

@@ -31,6 +31,10 @@
 _lock = threading.Lock()
 
 
+class PatStoreFormatError(ValueError):
+    """Raised when the PAT store JSON does not match its list-of-objects contract."""
+
+
 def ensure_pats_file() -> None:
     """Create the PATs file with an empty list if it doesn't exist. Called on validator boot."""
     with _lock:
@@ -49,7 +53,7 @@ def load_all_pats() -> list[dict]:
     with _lock:
         try:
             return _read_file()
-        except (json.JSONDecodeError, OSError) as e:
+        except (json.JSONDecodeError, OSError, PatStoreFormatError) as e:
             bt.logging.error(
                 f'miner_pats.json unreadable this round; scoring with no stored PATs until it recovers: {e}'
             )
@@ -102,13 +106,16 @@ def _read_file() -> list[dict]:
     """Read and parse the JSON store. Must be called while holding _lock.
 
     Returns [] only when the file genuinely does not exist. Raises
-    (json.JSONDecodeError / OSError) on a corrupt or unreadable file so the write
+    (ValueError / OSError) on a corrupt, structurally invalid, or unreadable file so the write
     path never mistakes a failed read for an empty store and overwrites it. The
     read paths that must tolerate a transient failure (load_all_pats) catch it.
     """
     if not PATS_FILE.exists():
         return []
-    return json.loads(PATS_FILE.read_text())
+    entries = json.loads(PATS_FILE.read_text())
+    if not isinstance(entries, list) or not all(isinstance(entry, dict) for entry in entries):
+        raise PatStoreFormatError('miner_pats.json must contain a list of objects')
+    return entries
 
 
 def _write_file(entries: list[dict]) -> None:

tests/validator/test_pat_handler.py

@@ -209,6 +209,20 @@ def test_unreadable_store_rejected_not_wiped(
         # The store was not overwritten down to the single incoming entry.
         assert use_tmp_pats_file.read_text() == 'not json{{{'
 
+    @patch('gittensor.validator.pat_handler._test_pat_against_repo', return_value=None)
+    @patch('gittensor.validator.pat_handler.validate_github_credentials', return_value=('github_42', None))
+    def test_invalid_store_schema_rejected_not_wiped(
+        self, mock_validate, mock_test_query, mock_validator, use_tmp_pats_file
+    ):
+        use_tmp_pats_file.write_text('null')
+
+        synapse = _make_broadcast_synapse('hotkey_1', pat='ghp_valid')
+        result = _run(handle_pat_broadcast(mock_validator, synapse))
+
+        assert result.accepted is False
+        assert 'temporarily unavailable' in (result.rejection_reason or '')
+        assert use_tmp_pats_file.read_text() == 'null'
+
     @patch('gittensor.validator.pat_handler.validate_github_credentials', return_value=(None, 'PAT invalid'))
     def test_invalid_pat_rejected(self, mock_validate, mock_validator):
         synapse = _make_broadcast_synapse('hotkey_1', pat='ghp_bad')
@@ -344,6 +358,16 @@ def test_unreadable_store_reports_inconclusive(self, mock_validator, use_tmp_pat
         assert result.pat_valid is None
         assert 'temporarily unavailable' in (result.rejection_reason or '')
 
+    def test_invalid_store_schema_reports_inconclusive(self, mock_validator, use_tmp_pats_file):
+        use_tmp_pats_file.write_text('null')
+
+        synapse = _make_check_synapse('hotkey_1')
+        result = _run(handle_pat_check(mock_validator, synapse))
+
+        assert result.has_pat is False
+        assert result.pat_valid is None
+        assert 'temporarily unavailable' in (result.rejection_reason or '')
+
     @patch('gittensor.validator.pat_handler._test_pat_against_repo')
     @patch(
         'gittensor.validator.utils.github_validation.get_github_identity',

tests/validator/test_pat_storage.py

@@ -91,6 +91,22 @@ def test_load_handles_corrupt_file(self, use_tmp_pats_file):
         entries = pat_storage.load_all_pats()
         assert entries == []
 
+    @pytest.mark.parametrize(
+        'payload',
+        [
+            'null',
+            '{}',
+            '"not a list"',
+            '[1]',
+            '["not an entry"]',
+            '[{"uid": 1}, null]',
+        ],
+    )
+    def test_load_handles_invalid_store_schema(self, use_tmp_pats_file, payload):
+        use_tmp_pats_file.write_text(payload)
+
+        assert pat_storage.load_all_pats() == []
+
 
 class TestSavePatFailsClosed:
     """The read-then-overwrite wipe (issue #1481, Proof 2): a single failed read of
@@ -135,6 +151,16 @@ def test_corrupt_store_save_does_not_shrink(self, use_tmp_pats_file):
 
         assert use_tmp_pats_file.read_text() == 'not json{{{'
 
+    @pytest.mark.parametrize('payload', ['null', '{}', '["not an entry"]'])
+    def test_invalid_store_schema_save_does_not_overwrite(self, use_tmp_pats_file, payload):
+        """Structurally invalid JSON must fail closed just like malformed JSON."""
+        use_tmp_pats_file.write_text(payload)
+
+        with pytest.raises(pat_storage.PatStoreFormatError, match='list of objects'):
+            pat_storage.save_pat(99, 'h99', 'p99', 'u99')
+
+        assert use_tmp_pats_file.read_text() == payload
+
 
 class TestGetPatByUid:
     def test_get_existing(self):