F20 Validate Genesis Stub

bottube_mood_engine.py

@@ -946,6 +946,12 @@ def _mood_service_unavailable(operation: str):
     return jsonify({"error": "Mood service unavailable"}), 500
 
 
+def _require_valid_agent_name(agent_name: str) -> Optional[tuple]:
+    if len(agent_name) > 128:
+        return jsonify({"error": "agent_name too long"}), 400
+    return None
+
+
 @mood_bp.route("/<agent_name>/mood", methods=["GET"])
 def get_agent_mood_endpoint(agent_name: str):
     """
@@ -956,6 +962,9 @@ def get_agent_mood_endpoint(agent_name: str):
     Query Parameters:
         include_stats - Include mood statistics (default: false)
     """
+    err = _require_valid_agent_name(agent_name)
+    if err:
+        return err
     try:
         engine = get_mood_engine()
         include_stats = request.args.get("include_stats", "false").lower() == "true"
@@ -984,6 +993,9 @@ def record_mood_signal(agent_name: str):
         value - Signal value data
         weight - Optional signal weight (default: 1.0)
     """
+    err = _require_valid_agent_name(agent_name)
+    if err:
+        return err
     try:
         engine = get_mood_engine()
         data, error = _get_json_object()
@@ -1014,6 +1026,9 @@ def generate_mood_title(agent_name: str):
     Request Body:
         topic - Video topic/theme
     """
+    err = _require_valid_agent_name(agent_name)
+    if err:
+        return err
     try:
         engine = get_mood_engine()
         data, error = _get_json_object()
@@ -1044,6 +1059,9 @@ def generate_mood_comment(agent_name: str):
     Request Body:
         base_comment - Optional base comment to modify
     """
+    err = _require_valid_agent_name(agent_name)
+    if err:
+        return err
     try:
         engine = get_mood_engine()
         data, error = _get_json_object(allow_empty=True)
@@ -1070,6 +1088,9 @@ def get_post_probability(agent_name: str):
 
     Get probability of agent posting based on current mood.
     """
+    err = _require_valid_agent_name(agent_name)
+    if err:
+        return err
     try:
         engine = get_mood_engine()
         probability = engine.get_post_probability(agent_name)
@@ -1093,6 +1114,9 @@ def get_mood_statistics_endpoint(agent_name: str):
 
     Get mood statistics for an agent.
     """
+    err = _require_valid_agent_name(agent_name)
+    if err:
+        return err
     try:
         engine = get_mood_engine()
         stats = engine.get_mood_statistics(agent_name)

bridge/bridge_api.py

@@ -194,7 +194,7 @@ def _json_object_body():
     return data, None
 
 
-def _clean_string_field(data, field_name, *, optional=False, lower=False):
+def _clean_string_field(data, field_name, *, optional=False, lower=False, max_length=0):
     value = data.get(field_name)
     if value is None:
         return None if optional else ""
@@ -205,6 +205,8 @@ def _clean_string_field(data, field_name, *, optional=False, lower=False):
         value = value.lower()
     if optional and not value:
         return None
+    if max_length > 0 and len(value) > max_length:
+        raise ValueError(f"{field_name} exceeds maximum length of {max_length}")
     return value
 
 
@@ -242,11 +244,11 @@ def lock_rtc():
 
     # ── Validate inputs ──
     try:
-        sender = _clean_string_field(data, "sender_wallet")
+        sender = _clean_string_field(data, "sender_wallet", max_length=128)
         target_chain = _clean_string_field(data, "target_chain", lower=True)
-        target_wallet = _clean_string_field(data, "target_wallet")
-        tx_hash = _clean_string_field(data, "tx_hash", optional=True)
-        receipt_signature = _clean_string_field(data, "receipt_signature", optional=True, lower=True)
+        target_wallet = _clean_string_field(data, "target_wallet", max_length=128)
+        tx_hash = _clean_string_field(data, "tx_hash", optional=True, max_length=128)
+        receipt_signature = _clean_string_field(data, "receipt_signature", optional=True, lower=True, max_length=256)
     except ValueError as exc:
         return jsonify({"error": str(exc)}), 400
 
@@ -394,8 +396,8 @@ def confirm_lock():
         return error_response
     try:
         lock_id = _clean_string_field(data, "lock_id")
-        proof_ref = _clean_string_field(data, "proof_ref")
-        notes = _clean_string_field(data, "notes", optional=True)
+        proof_ref = _clean_string_field(data, "proof_ref", max_length=256)
+        notes = _clean_string_field(data, "notes", optional=True, max_length=1024)
     except ValueError as exc:
         return jsonify({"error": str(exc)}), 400
 
@@ -460,8 +462,8 @@ def release_wrtc():
         return error_response
     try:
         lock_id = _clean_string_field(data, "lock_id")
-        release_tx = _clean_string_field(data, "release_tx")
-        notes = _clean_string_field(data, "notes", optional=True)
+        release_tx = _clean_string_field(data, "release_tx", max_length=128)
+        notes = _clean_string_field(data, "notes", optional=True, max_length=1024)
     except ValueError as exc:
         return jsonify({"error": str(exc)}), 400
 

contributor_registry.py

@@ -314,6 +314,8 @@ def api_contributors():
 def approve_contributor(username):
     if not _contributor_admin_authorized():
         abort(401)
+    if len(username) > 128:
+        abort(400, description="Username too long")
 
     with db_connection() as conn:
         conn.execute(

explorer/app.py

@@ -102,10 +102,14 @@ def get_network_stats():
 
 @app.route('/miner/<miner_id>')
 def miner_detail(miner_id):
+    if len(miner_id) > 128:
+        return render_template('404.html'), 404
     return render_template('miner_detail.html', miner_id=miner_id)
 
 @app.route('/api/miner/<miner_id>')
 def get_miner_detail(miner_id):
+    if len(miner_id) > 128:
+        return jsonify({'error': 'Miner not found'}), 404
     try:
         response = requests.get(MINERS_ENDPOINT, timeout=5)
         if response.status_code == 200:

explorer/rustchain_dashboard.py

@@ -746,6 +746,8 @@ def api_stats():
 @app.route('/api/wallet/<wallet_address>')
 def api_wallet_lookup(wallet_address):
     """Look up wallet balance and info"""
+    if len(wallet_address) > 128:
+        return jsonify({"error": "Invalid wallet address", "balance": 0}), 400
     try:
         with sqlite3.connect(DB_PATH) as conn:
             # Get balance

faucet.py

@@ -388,6 +388,9 @@ def drip():
 
     wallet = wallet_value.strip()
 
+    if len(wallet) > 128:
+        return jsonify({'ok': False, 'error': 'Wallet address too long'}), 400
+
     # Basic wallet validation (accept Ethereum-style and native RTC wallets)
     if not is_valid_wallet_address(wallet):
         return jsonify({'ok': False, 'error': 'Invalid wallet address'}), 400

health-dashboard/server.py

@@ -455,6 +455,8 @@ def api_status():
 @app.route('/api/history/<node_id>')
 def api_history(node_id: str):
     """API endpoint for historical data (24 hours)"""
+    if len(node_id) > 128:
+        return jsonify({"error": "Invalid node_id", "history": []}), 400
     conn = sqlite3.connect(DB_PATH)
     conn.row_factory = sqlite3.Row
     cursor = conn.cursor()

node/beacon_x402.py

@@ -104,13 +104,16 @@ def _json_object_body():
     return data, None
 
 
-def _json_string_field(data, field_name, default=""):
+def _json_string_field(data, field_name, default="", max_length=0):
     value = data.get(field_name, default)
     if value is None:
         return ""
     if not isinstance(value, str):
         raise ValueError(f"{field_name} must be a string")
-    return value.strip()
+    value = value.strip()
+    if max_length > 0 and len(value) > max_length:
+        raise ValueError(f"{field_name} exceeds maximum length of {max_length}")
+    return value
 
 
 def _is_base_address(value: str) -> bool:
@@ -211,6 +214,9 @@ def set_agent_wallet(agent_id):
         if request.method == "OPTIONS":
             return _cors_json({"ok": True})
 
+        if len(agent_id) > 128:
+            return _cors_json({"error": "agent_id too long"}, 400)
+
         # Simple admin check ? require admin key in header
         admin_error = _require_beacon_admin()
         if admin_error:
@@ -248,6 +254,9 @@ def get_agent_wallet(agent_id):
         if request.method == "OPTIONS":
             return _cors_json({"ok": True})
 
+        if len(agent_id) > 128:
+            return _cors_json({"error": "agent_id too long"}, 400)
+
         db = get_db_func()
 
         # Check beacon_wallets table (native agents)

node/bottube_embed.py

@@ -816,6 +816,8 @@ def embed_player(video_id: str):
     Returns:
         HTML page with embedded video player
     """
+    if len(video_id) > 256:
+        return Response("<html><body><h1>Invalid video ID</h1></body></html>", status=400, mimetype="text/html")
     # Get video data
     video = _get_mock_video(video_id)
 
@@ -861,6 +863,8 @@ def oembed():
         JSON oEmbed response
     """
     url = request.args.get("url", "")
+    if len(url) > 2048:
+        return jsonify({"error": "URL too long"}), 400
     format_param = request.args.get("format", "json")
     maxwidth = request.args.get("maxwidth", 854)
     maxheight = request.args.get("maxheight", 480)
@@ -955,6 +959,8 @@ def watch_page(video_id: str):
     Returns:
         Full HTML watch page
     """
+    if len(video_id) > 256:
+        return Response("<html><body><h1>Invalid video ID</h1></body></html>", status=400, mimetype="text/html")
     # Get video data
     video = _get_mock_video(video_id)
 

node/hall_of_rust.py

@@ -165,6 +165,8 @@ def induct_machine():
     # SECURITY FIX: Fingerprint based on HARDWARE ONLY (not wallet ID)
     # This prevents multiple wallets on same machine from getting multiple Hall entries
     hw_serial = data.get('cpu_serial', data.get('hardware_id', 'unknown'))
+    if not isinstance(hw_serial, str) or len(hw_serial) > 256:
+        hw_serial = 'unknown'
     fp_data = f"{data.get('device_model', '')}{data.get('device_arch', '')}{hw_serial}"
     fingerprint_hash = hashlib.sha256(fp_data.encode()).hexdigest()[:32]
 
@@ -180,8 +182,15 @@ def induct_machine():
         existing = c.fetchone()
 
         now = int(time.time())
-        model = data.get('device_model', 'Unknown')
-        arch = data.get('device_arch', 'modern')
+        miner_id = (data.get('miner_id') or 'anonymous')[:128]
+        model = (data.get('device_model', 'Unknown') or 'Unknown')[:256]
+        arch = (data.get('device_arch', 'modern') or 'modern')[:32]
+        device_family = (data.get('device_family', 'Unknown') or 'Unknown')[:128]
+
+        # Use defaults after truncation if empty
+        model = model or 'Unknown'
+        arch = arch or 'modern'
+        device_family = device_family or 'Unknown'
 
         if existing:
             # Update attestation count

node/machine_passport_api.py

@@ -284,6 +284,14 @@ def create_passport():
                 'message': "Field 'machine_id' is required or provide 'hardware_fingerprint'",
             }), 400
 
+    # Cap string fields to prevent DB abuse
+    name = (data.get('name') or '')[:256]
+    owner_miner_id = (data.get('owner_miner_id') or '')[:128]
+    architecture = (data.get('architecture') or '')[:64] if data.get('architecture') else None
+    photo_hash = (data.get('photo_hash') or '')[:128] if data.get('photo_hash') else None
+    photo_url = (data.get('photo_url') or '')[:2048] if data.get('photo_url') else None
+    provenance = (data.get('provenance') or '')[:1024] if data.get('provenance') else None
+
     ledger = get_ledger()
 
     # Check if passport already exists
@@ -297,13 +305,13 @@ def create_passport():
 
     passport = MachinePassport(
         machine_id=machine_id,
-        name=data['name'],
-        owner_miner_id=data['owner_miner_id'],
+        name=name,
+        owner_miner_id=owner_miner_id,
         manufacture_year=data.get('manufacture_year'),
-        architecture=data.get('architecture'),
-        photo_hash=data.get('photo_hash'),
-        photo_url=data.get('photo_url'),
-        provenance=data.get('provenance'),
+        architecture=architecture,
+        photo_hash=photo_hash,
+        photo_url=photo_url,
+        provenance=provenance,
     )
 
     success, msg = ledger.create_passport(passport)
@@ -350,6 +358,13 @@ def update_passport(machine_id: str):
             'message': 'JSON body required',
         }), 400
 
+    # Cap string fields to prevent DB abuse
+    for field in ('name', 'owner_miner_id', 'architecture', 'photo_hash', 'photo_url', 'provenance', 'machine_id'):
+        if field in data and isinstance(data[field], str):
+            max_len = {'machine_id': 128, 'name': 256, 'owner_miner_id': 128, 'architecture': 64,
+                       'photo_hash': 128, 'photo_url': 2048, 'provenance': 1024}.get(field, 256)
+            data[field] = data[field][:max_len]
+
     success, msg = ledger.update_passport(machine_id, data)
 
     if success:

node/rustchain_p2p_sync.py

@@ -446,6 +446,9 @@ def announce_peer():
 
         peer_url = peer_url.strip()
 
+        if len(peer_url) > 1024:
+            return jsonify({"ok": False, "error": "peer_url too long"}), 400
+
         if peer_url:
             success = peer_manager.add_peer(peer_url)
             return jsonify({"ok": success, "peers": len(peer_manager.get_active_peers())})