fix: security fixes

compute_horde/compute_horde/miner_client/organic.py

@@ -56,6 +56,7 @@
     JobAcceptedReceiptPayload,
     JobFinishedReceiptPayload,
     JobStartedReceiptPayload,
+    clamp_ttl,
 )
 from compute_horde.transport import (
     AbstractTransport,
@@ -107,11 +108,11 @@ class OrganicMinerClient(AbstractMinerClient[MinerToValidatorMessage, ValidatorT
         client = OrganicMinerClient(...)
         async with client:
             try:
-                await client.send_model(V0InitialJobRequest(...))
+                await client.send_model(V0InitialJobRequest(...).signed(client.my_keypair))
                 msg = await client.job_accepted_future
                 # handle msg
 
-                await client.send_model(V0JobRequest(...))
+                await client.send_model(V0JobRequest(...).signed(client.my_keypair))
                 msg = client.job_finished_future
                 # handle msg
             except MinerRejectedJob:
@@ -160,7 +161,9 @@ def __init__(
         self.miner_machine_specs: MachineSpecs | None = None
 
         name = f"{miner_hotkey}({miner_address}:{miner_port})"
-        transport = transport or WSTransport(name, self.miner_url())
+        transport = transport or WSTransport(
+            name, self.miner_url(), reconnect_greeting=self._serialized_authentication_message
+        )
         super().__init__(name, transport)
 
     @cached_property
@@ -344,6 +347,9 @@ def generate_authentication_message(self) -> ValidatorAuthForMiner:
         msg.signature = sign_blob(self.my_keypair, msg.blob_for_signing())
         return msg
 
+    def _serialized_authentication_message(self) -> str:
+        return self.generate_authentication_message().model_dump_json()
+
     def generate_job_started_receipt_message(
         self,
         executor_class: ExecutorClass,
@@ -373,7 +379,7 @@ def generate_job_accepted_receipt_message(
             validator_hotkey=self.my_hotkey,
             timestamp=datetime.datetime.now(datetime.UTC),
             time_accepted=time_accepted,
-            ttl=ttl,
+            ttl=clamp_ttl(ttl),
         )
         return V0JobAcceptedReceiptRequest(
             payload=receipt_payload,
@@ -415,7 +421,8 @@ def generate_job_finished_receipt_message(
             validator_hotkey=self.my_hotkey,
             timestamp=datetime.datetime.now(datetime.UTC),
             time_started=time_started,
-            time_took_us=int(time_took_seconds * 1_000_000),
+            # Wall-clock durations go negative after a backwards clock step; the payload rejects that.
+            time_took_us=max(0, int(time_took_seconds * 1_000_000)),
             block_numbers=allowance_blocks,
             score_str=f"{score:.6f}",
         )
@@ -471,7 +478,7 @@ async def send_initial_job_request(
                 if job_details.job_timing is not None
                 else None,
                 streaming_details=job_details.streaming_details,
-            ),
+            ).signed(self.my_keypair),
         )
 
     async def send_model(
@@ -484,7 +491,8 @@ async def send_model(
 
     async def connect(self) -> None:
         await super().connect()
-        await self.transport.send(self.generate_authentication_message().model_dump_json())
+        # The default WSTransport authenticates every reopened connection via reconnect_greeting.
+        await self.transport.send(self._serialized_authentication_message())
 
 
 @dataclass
@@ -619,7 +627,7 @@ async def execute_organic_job_on_miner(
                     volume=None,  # Was sent in the initial request
                     output_upload=job_details.output,
                     artifacts_dir=job_details.artifacts_dir,
-                )
+                ).signed(client.my_keypair)
             )
 
             ## STAGE: Volume download

compute_horde/compute_horde/protocol_consts.py

@@ -6,6 +6,10 @@
 else:
     from backports.strenum import StrEnum  #  noqa: UP035
 
+# Miner hotkey that a validator puts in the authentication frames and receipts it sends to its
+# trusted miner, which it reaches by address rather than by a hotkey registered on the subnet.
+TRUSTED_MINER_FAKE_KEY = "0" * 48
+
 
 class JobParticipantType(StrEnum):
     UNKNOWN = "unknown"
@@ -151,6 +155,7 @@ class JobRejectionReason(StrEnum):
     UNKNOWN = "unknown"
     BUSY = "busy"
     INVALID_SIGNATURE = "invalid_signature"
+    INVALID_RECEIPT = "invalid_receipt"
     NO_MINER_FOR_JOB = "no_miner_for_job"
     VALIDATOR_BLACKLISTED = "validator_blacklisted"
 

compute_horde/compute_horde/protocol_messages.py

@@ -1,11 +1,13 @@
 import enum
-from typing import Annotated, Literal, TypeAlias
+import json
+from typing import Annotated, Any, Literal, Self, TypeAlias
 
+import bittensor
 from compute_horde_core.executor_class import ExecutorClass
 from compute_horde_core.output_upload import OutputUpload
 from compute_horde_core.streaming import StreamingDetails
 from compute_horde_core.volume import Volume
-from pydantic import AliasChoices, BaseModel, Field, JsonValue
+from pydantic import AliasChoices, BaseModel, Field, JsonValue, PrivateAttr, TypeAdapter
 from typing_extensions import deprecated
 
 from compute_horde.base.docker import DockerRunOptionsPreset
@@ -22,7 +24,7 @@
     JobStartedReceiptPayload,
     Receipt,
 )
-from compute_horde.utils import MachineSpecs
+from compute_horde.utils import MachineSpecs, canonical_json, sign_blob, verify_blob_signature
 
 # NOTE:
 # miner.ec = Miner's executor consumer
@@ -78,8 +80,43 @@ class V0MainHotkeyMessage(BaseModel):
     main_hotkey: str | None = None
 
 
+# validator -> miner.vc
+class ValidatorSignedMessage(BaseModel):
+    """
+    Message whose whole content is signed with the validator hotkey, so the miner can tell a
+    request authored by the authenticated validator from one injected into its socket.
+
+    The signature covers the message's JSON object without validator_signature, in canonical
+    form. A message from parse_validator_to_miner_message is verified against the object as it
+    arrived rather than against its own fields, so a field known to only one side keeps the
+    signature valid. Copies keep that object too, so verify a received message before changing
+    it.
+    """
+
+    validator_signature: str | None = None
+    _received_content: dict[str, Any] | None = PrivateAttr(default=None)
+
+    def blob_for_signing(self) -> str:
+        content = self._received_content
+        if content is None:
+            content = json.loads(self.model_dump_json(exclude={"validator_signature"}))
+        return canonical_json(content)
+
+    def signed(self, keypair: bittensor.Keypair) -> Self:
+        return self.model_copy(
+            update={"validator_signature": sign_blob(keypair, self.blob_for_signing())}
+        )
+
+    def is_signed_by(self, validator_hotkey: str) -> bool:
+        if self.validator_signature is None:
+            return False
+        return verify_blob_signature(
+            validator_hotkey, self.blob_for_signing(), self.validator_signature
+        )
+
+
 # validator -> miner.vc -> miner.ec -> executor
-class V0InitialJobRequest(BaseModel):
+class V0InitialJobRequest(ValidatorSignedMessage):
     class ExecutorTimingDetails(BaseModel):
         allowed_leeway: int
         download_time_limit: int
@@ -172,7 +209,7 @@ class V0ExecutionDoneRequest(BaseModel):
 
 
 # validator -> miner.vc -> miner.ec -> executor
-class V0JobRequest(BaseModel):
+class V0JobRequest(ValidatorSignedMessage):
     message_type: Literal["V0JobRequest"] = "V0JobRequest"
     job_uuid: str
     executor_class: ExecutorClass
@@ -264,6 +301,32 @@ class V0HordeFailedRequest(BaseModel):
     Field(discriminator="message_type"),
 ]
 
+_validator_to_miner_message: TypeAdapter[ValidatorToMinerMessage] = TypeAdapter(
+    ValidatorToMinerMessage
+)
+
+
+def parse_validator_to_miner_message(raw: str | bytes) -> ValidatorToMinerMessage:
+    """
+    Parses a frame sent by a validator. A signed message is validated from the same JSON object
+    that its signature is checked against, so with duplicate keys, too, the miner acts only on
+    content the validator signed.
+
+    Raises ValueError, such as json.JSONDecodeError or pydantic.ValidationError, for a malformed
+    frame.
+    """
+    try:
+        content = json.loads(raw)
+        message = _validator_to_miner_message.validate_python(content)
+    except RecursionError as ex:
+        raise ValueError("JSON is nested too deeply") from ex
+    if isinstance(message, ValidatorSignedMessage):
+        message._received_content = {
+            key: value for key, value in content.items() if key != "validator_signature"
+        }
+    return message
+
+
 MinerToExecutorMessage = Annotated[
     GenericError | V0InitialJobRequest | V0JobRequest,
     Field(discriminator="message_type"),

compute_horde/compute_horde/receipts/models.py

@@ -8,10 +8,12 @@
 from compute_horde.executor_class import DEFAULT_EXECUTOR_CLASS
 from compute_horde.receipts import ReceiptType
 from compute_horde.receipts.schemas import (
+    SCORE_STR_PATTERN,
     JobAcceptedReceiptPayload,
     JobFinishedReceiptPayload,
     JobStartedReceiptPayload,
     Receipt,
+    parse_score_str,
 )
 
 
@@ -181,8 +183,18 @@ class JobFinishedReceipt(AbstractReceipt):
     def time_took(self):
         return timedelta(microseconds=self.time_took_us)
 
-    def score(self):
-        return float(self.score_str)
+    def score(self) -> float:
+        """
+        Raises ValueError for a persisted row whose score_str predates wire-format validation.
+        """
+        return parse_score_str(self.score_str)
+
+    @classmethod
+    def with_malformed_score(cls) -> models.QuerySet["JobFinishedReceipt"]:
+        """
+        Rows for which score() raises ValueError.
+        """
+        return cls.objects.exclude(score_str__regex=SCORE_STR_PATTERN.pattern)
 
     def to_receipt(self) -> Receipt:
         if self.miner_signature is None: