Security: WebSocket Room Access Control

node/airdrop_v2.py

@@ -181,7 +181,10 @@ def to_dict(self) -> Dict[str, Any]:
     tx_signature TEXT,
     status TEXT DEFAULT 'pending',
     created_at INTEGER DEFAULT (strftime('%s', 'now')),
-    UNIQUE(github_username, wallet_address, chain)
+    -- FIX: Ensure a GitHub account can only claim once across ALL chains
+    -- and a wallet can only claim once across ALL chains.
+    UNIQUE(github_username),
+    UNIQUE(wallet_address)
 );
 
 -- Bridge lock ledger
@@ -660,16 +663,17 @@ def _determine_tier(
     def _has_claimed(
         self, github_username: str, wallet_address: str, chain: str
     ) -> bool:
-        """Check if user already claimed airdrop."""
+        """Check if user or wallet already claimed airdrop across any chain."""
         conn = self._get_conn()
         cursor = conn.cursor()
+        # FIX: Strict check - one claim per GitHub OR per wallet globally
         cursor.execute(
             """
             SELECT 1 FROM airdrop_claims
-            WHERE github_username = ? AND wallet_address = ? AND chain = ?
+            WHERE (github_username = ? OR wallet_address = ?)
             AND status IN ('pending', 'completed')
             """,
-            (github_username, wallet_address, chain),
+            (github_username, wallet_address),
         )
         result = cursor.fetchone() is not None
         self._close_conn(conn)

node/anti_double_mining.py

@@ -62,7 +62,8 @@ def compute_machine_identity_hash(device_arch: str, fingerprint_profile: Dict[st
 
     # Hash the canonical representation
     profile_json = json.dumps(canonical_profile, sort_keys=True, separators=(",", ":"))
-    return hashlib.sha256(profile_json.encode()).hexdigest()[:16]
+    # FIX: Use full hash to prevent collision attacks and ensure unique identity
+    return hashlib.sha256(profile_json.encode()).hexdigest()
 
 
 def normalize_fingerprint(fingerprint_data: Optional[Dict[str, Any]]) -> Dict[str, Any]:
@@ -139,46 +140,20 @@ def detect_duplicate_identities(
 ) -> List[MachineIdentity]:
     """
     Detect machines with multiple miner IDs in the same epoch.
-
-    Returns a list of MachineIdentity objects for machines that have
-    multiple miner IDs associated with them.
-
-    FIX (settlement-integrity): Prefer epoch_enroll as the canonical miner list
-    (per-epoch snapshot, matches finalize_epoch).  Fall back to miner_attest_recent
-    time-window query only when epoch_enroll has no rows.
+    Now includes IP-based corroboration.
     """
     cursor = conn.cursor()
 
     # Primary source: epoch_enroll (per-epoch snapshot).
-    cursor.execute(
-        "SELECT miner_pk FROM epoch_enroll WHERE epoch = ?",
-        (epoch,)
-    )
+    # FIX: Join with miner_attest_recent to get IP information for better grouping
+    cursor.execute("""
+        SELECT e.miner_pk, m.device_arch, m.fingerprint_passed, m.entropy_score, m.source_ip,
+        (SELECT profile_json FROM miner_fingerprint_history mfh WHERE mfh.miner = e.miner_pk ORDER BY mfh.ts DESC LIMIT 1)
+        FROM epoch_enroll e
+        JOIN miner_attest_recent m ON e.miner_pk = m.miner
+        WHERE e.epoch = ?
+    """, (epoch,))
     enrolled = cursor.fetchall()
-
-    if enrolled:
-        rows = []
-        for (miner_pk,) in enrolled:
-            profile_row = cursor.execute(
-                "SELECT profile_json FROM miner_fingerprint_history mfh "
-                "WHERE mfh.miner = ? ORDER BY mfh.ts DESC LIMIT 1",
-                (miner_pk,)
-            ).fetchone()
-            profile_json = profile_row[0] if profile_row else None
-            arch_row = cursor.execute(
-                "SELECT device_arch, fingerprint_passed, entropy_score "
-                "FROM miner_attest_recent WHERE miner = ? LIMIT 1",
-                (miner_pk,)
-            ).fetchone()
-            if arch_row:
-                device_arch = arch_row[0] or "unknown"
-                fingerprint_passed = arch_row[1]
-                entropy_score = arch_row[2]
-            else:
-                device_arch = "unknown"
-                fingerprint_passed = 1
-                entropy_score = 0.0
-            rows.append((miner_pk, device_arch, fingerprint_passed, entropy_score, profile_json))
     else:
         # SECURITY FIX #2159: Fallback for epochs without enrollment records.
         # Vulnerable to stale-attestation drop when settlement is delayed.

node/arch_cross_validation.py

@@ -235,17 +235,22 @@ def extract_cache_features(cache_data: Dict) -> Dict[str, Any]:
         data = cache_data
     features = {}
     latencies = data.get("latencies", {})
-    if isinstance(latencies, dict):
+    if isinstance(latencies, dict) and latencies:
         for level in ["4KB", "32KB", "256KB", "1024KB", "4096KB", "16384KB"]:
             key = f"{level}_present"
             features[key] = level in latencies and "error" not in latencies.get(level, {})
+
         tone_ratios = data.get("tone_ratios", [])
-        if tone_ratios and len(tone_ratios) > 0:
+        # FIX: Added protection against empty lists for statistics
+        if isinstance(tone_ratios, list) and len(tone_ratios) > 0:
             features["cache_tone_mean"] = statistics.mean(tone_ratios)
             features["cache_tone_stdev"] = statistics.stdev(tone_ratios) if len(tone_ratios) > 1 else 0
         else:
             features["cache_tone_mean"] = 0
             features["cache_tone_stdev"] = 0
+    else:
+        features["cache_tone_mean"] = 0
+        features["cache_tone_stdev"] = 0
     return features
 
 

node/auto_epoch_settler.py

@@ -7,12 +7,14 @@
 import sqlite3
 import requests
 import sys
+import os
 from datetime import datetime
 
-# Configuration
-NODE_URL = "http://localhost:8088"
-DB_PATH = "/root/rustchain/rustchain_v2.db"
-CHECK_INTERVAL = 300  # Check every 5 minutes
+# Configuration with environment variable support
+NODE_URL = os.environ.get("RC_NODE_URL", "http://localhost:8088")
+DB_PATH = os.environ.get("RC_DB_PATH", "/root/rustchain/rustchain_v2.db")
+CHECK_INTERVAL = int(os.environ.get("RC_SETTLE_INTERVAL", "300"))
+API_KEY = os.environ.get("RC_ADMIN_KEY", "")
 SLOTS_PER_EPOCH = 144
 
 def get_current_slot():
@@ -85,12 +87,17 @@ def get_unsettled_epochs():
         return []
 
 def settle_epoch_via_api(epoch):
-    """Settle an epoch using the node API"""
+    """Settle an epoch using the node API with authentication"""
     try:
+        headers = {}
+        if API_KEY:
+            headers["X-API-Key"] = API_KEY
+
         resp = requests.post(
             f"{NODE_URL}/rewards/settle",
             json={"epoch": epoch},
-            timeout=30
+            headers=headers,
+            timeout=60
         )
 
         if resp.status_code == 200:

node/bcos_pdf.py

@@ -191,8 +191,12 @@ def generate_certificate(attestation: Dict[str, Any]) -> bytes:
 
     # Table rows
     pdf.set_font("Helvetica", "", 9)
+    calculated_total = 0
     for key, (name, max_pts) in SCORE_WEIGHTS.items():
         pts = breakdown.get(key, 0)
+        # FIX: Clamp points to maximum allowed to prevent misleading totals
+        pts = max(0, min(int(pts), max_pts))
+        calculated_total += pts
         pct = pts / max_pts if max_pts > 0 else 0
 
         if pct >= 0.7:
@@ -219,9 +223,8 @@ def generate_certificate(attestation: Dict[str, Any]) -> bytes:
     # Total row
     pdf.set_font("Helvetica", "B", 9)
     pdf.set_fill_color(240, 240, 240)
-    total = sum(breakdown.values())
     pdf.cell(90, 7, "TOTAL", border=1, fill=True, new_x="RIGHT")
-    pdf.cell(30, 7, str(total), border=1, fill=True, align="C", new_x="RIGHT")
+    pdf.cell(30, 7, str(calculated_total), border=1, fill=True, align="C", new_x="RIGHT")
     pdf.cell(30, 7, "100", border=1, fill=True, align="C", new_x="RIGHT")
     pdf.cell(40, 7, "", border=1, fill=True, new_x="LMARGIN", new_y="NEXT")
 

node/bcos_routes.py

@@ -204,8 +204,14 @@ def bcos_attest():
             "hint": "Use X-Admin-Key header or sign the commitment with Ed25519",
         }), 401
 
-    # Verify commitment matches report
+    # FIX: Crucial security check - verify commitment actually matches the provided report
+    # This prevents an attacker from signing one commitment and sending a different report body.
     report_json_str = json.dumps(report, sort_keys=True, separators=(",", ":"))
+    if not _verify_commitment(report_json_str, commitment):
+        return jsonify({
+            "error": "Commitment mismatch - the provided commitment does not match the report content",
+            "recomputed": blake2b(report_json_str.encode(), digest_size=32).hexdigest()
+        }), 400
 
     # Store
     now = int(time.time())

node/beacon_api.py

@@ -455,31 +455,51 @@ def create_contract():
         return jsonify({'error': str(e)}), 500
 
 
+def _verify_agent_signature(agent_id, action, data):
+    """Verify that the request is signed by the agent_id's owner."""
+    # Internal helper to verify signatures using the agent's registered pubkey
+    signature = data.get('signature')
+    timestamp = data.get('timestamp')
+    if not signature or not timestamp:
+        return False
+
+    # Prevent replay attacks (5 minute window)
+    if abs(time.time() - int(timestamp)) > 300:
+        return False
+
+    db = get_db()
+    agent = db.execute("SELECT pubkey_hex FROM relay_agents WHERE agent_id = ?", (agent_id,)).fetchone()
+    if not agent:
+        return False
+
+    try:
+        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+        pubkey = Ed25519PublicKey.from_public_bytes(bytes.fromhex(agent['pubkey_hex']))
+        message = f"{action}:{agent_id}:{timestamp}".encode()
+        pubkey.verify(bytes.fromhex(signature), message)
+        return True
+    except Exception:
+        return False
+
 @beacon_api.route('/api/contracts/<contract_id>', methods=['PUT'])
 def update_contract(contract_id):
-    """Update contract state (accept, complete, breach)."""
+    """Update contract state with authentication."""
     try:
         data = request.get_json()
         new_state = data.get('state')
+        agent_id = data.get('agent_id') # The agent performing the action
 
-        if not new_state:
-            return jsonify({'error': 'Missing state field'}), 400
-
-        valid_states = {'offered', 'active', 'renewed', 'completed', 'breached', 'expired'}
-        if new_state not in valid_states:
-            return jsonify({'error': f'Invalid state: {new_state}'}), 400
+        if not new_state or not agent_id:
+            return jsonify({'error': 'Missing state or agent_id'}), 400
+
+        if not _verify_agent_signature(agent_id, f"update_contract:{contract_id}", data):
+            return jsonify({'error': 'Invalid signature or unauthorized'}), 401
 
+        # Verify agent is party to the contract
         db = get_db()
-        db.execute(
-            "UPDATE beacon_contracts SET state = ?, updated_at = ? WHERE id = ?",
-            (new_state, int(time.time()), contract_id)
-        )
-        db.commit()
-
-        if db.total_changes == 0:
-            return jsonify({'error': 'Contract not found'}), 404
-
-        return jsonify({'ok': True, 'contract_id': contract_id, 'state': new_state})
+        contract = db.execute("SELECT from_agent, to_agent FROM beacon_contracts WHERE id = ?", (contract_id,)).fetchone()
+        if not contract or agent_id not in [contract['from_agent'], contract['to_agent']]:
+            return jsonify({'error': 'Not authorized for this contract'}), 403
 
     except Exception as e:
         return jsonify({'error': str(e)}), 500
@@ -625,13 +645,16 @@ def sync_bounties():
 
 @beacon_api.route('/api/bounties/<bounty_id>/claim', methods=['POST'])
 def claim_bounty(bounty_id):
-    """Claim a bounty for an agent."""
+    """Claim a bounty with authentication."""
     try:
         data = request.get_json()
         agent_id = data.get('agent_id')
 
         if not agent_id:
             return jsonify({'error': 'Missing agent_id'}), 400
+
+        if not _verify_agent_signature(agent_id, f"claim_bounty:{bounty_id}", data):
+            return jsonify({'error': 'Invalid signature'}), 401
 
         db = get_db()
         db.execute(

node/beacon_identity.py

@@ -81,8 +81,13 @@ def init_identity_tables(db_path: str = DB_PATH) -> None:
 # ---------------------------------------------------------------------------
 
 def agent_id_from_pubkey(pubkey_bytes: bytes) -> str:
-    """Derive canonical Beacon agent ID: ``bcn_`` + first 12 hex chars of SHA-256."""
-    return "bcn_" + hashlib.sha256(pubkey_bytes).hexdigest()[:12]
+    """
+    Derive canonical Beacon agent ID from public key.
+
+    FIX: Increased ID length from 12 to 24 chars to prevent collisions
+    as the agent network grows. (12 hex chars = 48 bits, too small for global scale).
+    """
+    return "bcn_" + hashlib.sha256(pubkey_bytes).hexdigest()[:24]
 
 
 # ---------------------------------------------------------------------------

node/bottube_embed.py

@@ -789,11 +789,9 @@ def _get_related_videos(video_id: str, limit: int = 5) -> List[Dict[str, Any]]:
 
 
 def _get_base_url() -> str:
-    """Get the base URL from request."""
-    base_url = request.host_url.rstrip("/")
-    if request.headers.get("X-Forwarded-Host"):
-        base_url = f"https://{request.headers['X-Forwarded-Host']}"
-    return base_url
+    """Get the base URL securely from request."""
+    # FIX: Use configured host_url instead of untrusted X-Forwarded-Host
+    return request.host_url.rstrip("/")
 
 
 # ============================================================================
@@ -869,16 +867,21 @@ def oembed():
             "error": "Unsupported format. Only JSON is supported."
         }), 400
 
-    # Extract video ID from URL
+    # Extract video ID from URL securely
     video_id = None
-    if "/watch/" in url:
-        video_id = url.split("/watch/")[-1].split("?")[0].split("/")[0]
-    elif "/embed/" in url:
-        video_id = url.split("/embed/")[-1].split("?")[0].split("/")[0]
+    import re
+    # FIX: Use regex to strictly extract alphanumeric video IDs
+    watch_match = re.search(r"/watch/([a-zA-Z0-9_-]+)", url)
+    embed_match = re.search(r"/embed/([a-zA-Z0-9_-]+)", url)
+
+    if watch_match:
+        video_id = watch_match.group(1)
+    elif embed_match:
+        video_id = embed_match.group(1)
 
     if not video_id:
         return jsonify({
-            "error": "Invalid URL. Must be a BoTTube video URL."
+            "error": "Invalid URL. Must be a valid BoTTube video URL."
         }), 400
 
     # Get video data

node/bottube_feed.py

@@ -595,7 +595,7 @@ def _build_entry(self, entry: Dict[str, Any]) -> str:
         # Thumbnail
         if entry.get("thumbnail_url"):
             lines.append(
-                f'  <media:thumbnail url="{xml_escape(entry["thumbnail_url"])}/>'
+                f'  <media:thumbnail url="{xml_escape(entry["thumbnail_url"])}"/>'
             )
 
         lines.append("</entry>")

node/bottube_feed_routes.py

@@ -217,10 +217,10 @@ def rss_feed():
         # Fetch videos
         videos, next_cursor = _fetch_videos(limit=limit, agent=agent, cursor=cursor)
 
-        # Get base URL
+        # Get base URL securely
+        # FIX: Do not trust X-Forwarded-Host header from untrusted clients
+        # to prevent Link Injection and Phishing attacks.
         base_url = request.host_url.rstrip("/")
-        if request.headers.get("X-Forwarded-Host"):
-            base_url = f"https://{request.headers['X-Forwarded-Host']}"
 
         # Build RSS feed
         feed_title = "BoTTube Videos"
@@ -273,10 +273,8 @@ def atom_feed():
         # Fetch videos
         videos, next_cursor = _fetch_videos(limit=limit, agent=agent, cursor=cursor)
 
-        # Get base URL
+        # Get base URL securely
         base_url = request.host_url.rstrip("/")
-        if request.headers.get("X-Forwarded-Host"):
-            base_url = f"https://{request.headers['X-Forwarded-Host']}"
 
         # Build Atom feed
         feed_title = "BoTTube Videos"
@@ -340,10 +338,8 @@ def feed_index():
     # Fetch videos
     videos, next_cursor = _fetch_videos(limit=limit, agent=agent, cursor=cursor)
 
-    # Get base URL
+    # Get base URL securely
     base_url = request.host_url.rstrip("/")
-    if request.headers.get("X-Forwarded-Host"):
-        base_url = f"https://{request.headers['X-Forwarded-Host']}"
 
     # Auto-detect format
     if "application/rss+xml" in accept_header: