Fix bandit issues

Author: cdonnachie

.bandit

@@ -0,0 +1,17 @@
+[bandit]
+# Configuration file for Bandit security scanner
+# See: https://bandit.readthedocs.io/en/latest/configuration/index.html
+
+exclude_dirs = ['/tests', '/test', '/.venv', '/venv', '.git']
+tests = 
+skips = 
+
+[bandit-assert_used:B101]
+# Skip assert checks - we use asserts only for input validation in non-critical paths
+skips = 
+
+[bandit-hardcoded_bind_all_interfaces:B104]
+# 0.0.0.0 is intentional for server services (stratum proxy, dashboard)
+# These are server-side network services, not client connections
+# Binding to all interfaces is the correct behavior for public-facing services
+skips = 

kcn_proxy/state/updater.py

@@ -1,4 +1,4 @@
-import time, os, base58
+import time, os, base58, logging
 from aiohttp import ClientSession
 from ..rpc import kcn as rpc_kcn
 from ..consensus.merkle import merkle_root_from_txids_le, merkle_branch_for_index0
@@ -8,6 +8,8 @@
 )
 from ..consensus.auxpow import refresh_aux_job
 
+logger = logging.getLogger(__name__)
+
 
 async def update_once(state, settings, http: ClientSession, force_update: bool = False):
     ROLL_SECONDS = getattr(settings, "ntime_roll", 30)
@@ -168,8 +170,8 @@ async def update_once(state, settings, http: ClientSession, force_update: bool =
                         from ..stratum.session import hashratedict
 
                         hashratedict.pop(wid, None)
-                except Exception:
-                    pass
+                except Exception as e:
+                    logger.debug("Failed to remove worker %s from hashrate tracker: %s", wid, e)
             else:
                 alive.add(sess)
         state.all_sessions = alive

kcn_proxy/stratum/server.py

@@ -55,8 +55,8 @@ async def start_server(state, settings):
             # Force an initial tick so the (empty) state file is created promptly
             try:
                 await _vardiff_mod.vardiff_manager.tick()
-            except Exception:
-                pass
+            except Exception as e:
+                logger.debug("Initial vardiff tick failed: %s", e)
 
             # Periodic tick task
             async def _vardiff_tick_loop():
@@ -66,8 +66,8 @@ async def _vardiff_tick_loop():
                     try:
                         await asyncio.sleep(30)
                         await _vardiff_mod.vardiff_manager.tick()
-                    except Exception:
-                        pass
+                    except Exception as e:
+                        logger.debug("Periodic vardiff tick failed: %s", e)
 
             import asyncio
 

kcn_proxy/stratum/session.py

@@ -394,8 +394,8 @@ async def connection_lost(self):
             wid = getattr(self, "_worker_id", None)
             if wid:
                 hashrate_tracker.remove_worker(wid)
-        except Exception:
-            pass
+        except Exception as e:
+            self.logger.debug("Failed to remove worker from hashrate tracker: %s", e)
 
         self._state.new_sessions.discard(self)
         self._state.all_sessions.discard(self)
@@ -405,8 +405,8 @@ async def connection_lost(self):
         except TypeError:
             try:
                 super().connection_lost()
-            except Exception:
-                pass
+            except Exception as e:
+                self.logger.debug("Error calling connection_lost: %s", e)
 
     async def handle_subscribe(self, *args):
         if self not in self._state.all_sessions:
@@ -570,8 +570,8 @@ async def _keepalive_loop(self):
                             if abs(vd - difficulty) / max(difficulty, 1e-9) >= 0.05:
                                 difficulty = vd
                                 self._share_difficulty = vd
-                        except Exception:
-                            pass
+                        except Exception as e:
+                            self.logger.debug("Vardiff adjustment failed: %s", e)
                     await self.send_notification("mining.set_difficulty", (difficulty,))
                     self._last_activity = loop.time()
                     self.logger.debug(
@@ -719,8 +719,8 @@ async def handle_submit(self, *args, **kwargs):
             # Record rejected share (confidence accounting) using assigned diff
             try:
                 hashrate_tracker.add_share(worker, sent_diff, accepted=False)
-            except Exception:
-                pass
+            except Exception as e:
+                self.logger.debug("Failed to record rejected share: %s", e)
             if self._debug_shares:
                 self.logger.error(
                     "Low difficulty share: shareDiff=%.18f minerDiff=%.18f",
@@ -744,8 +744,8 @@ async def handle_submit(self, *args, **kwargs):
                 await _vardiff_mod.vardiff_manager.record_share(
                     worker, share_difficulty=sent_diff
                 )
-            except Exception:
-                pass
+            except Exception as e:
+                self.logger.debug("Failed to record share for vardiff: %s", e)
 
         # Log share statistics asynchronously
         import time

kcn_proxy/stratum/vardiff.py

@@ -1,9 +1,12 @@
 import time
 import asyncio
+import logging
 from collections import deque
 from dataclasses import dataclass
 from typing import Deque, Dict, Optional
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass
 class MinerState:
@@ -156,8 +159,8 @@ def _maybe_retarget(self, st: MinerState):
                     cap = chain_diff * headroom
                     if new_diff > cap:
                         new_diff = cap
-        except Exception:
-            pass
+        except Exception as e:
+            logger.debug("Error adjusting difficulty for chain headroom: %s", e)
 
         # Apply only if material (>5%) change
         if abs(new_diff - st.difficulty) / max(st.difficulty, 1e-12) >= 0.05:
@@ -203,8 +206,8 @@ def _save_state(self):
             }
             with open(self.state_path, "w") as f:
                 json.dump(data, f)
-        except Exception:
-            pass
+        except Exception as e:
+            logger.debug("Failed to save vardiff state: %s", e)
 
     def _load_state(self):
         try:
@@ -223,8 +226,8 @@ def _load_state(self):
                     last_retarget=vd.get("last_retarget", now),
                     ema_interval=vd.get("ema_interval"),
                 )
-        except Exception:
-            pass
+        except Exception as e:
+            logger.debug("Failed to load vardiff state: %s", e)
 
     def export_state(self) -> dict:
         return {

kcn_proxy/utils/enc.py

@@ -2,7 +2,8 @@
 
 
 def var_int(i: int) -> bytes:
-    assert i >= 0
+    if i < 0:
+        raise ValueError(f"var_int requires non-negative integer, got {i}")
     if i < 0xFD:
         return i.to_bytes(1, "little")
     if i <= 0xFFFF:

kcn_proxy/web/api.py

@@ -559,8 +559,8 @@ async def lcn_hash_fix_status():
                         row = await cur.fetchone()
                         if row:
                             bad_hash_count = row[0]
-            except Exception:
-                pass  # DB might not exist or table not created
+            except Exception as e:
+                logger.debug("Failed to query bad hash count: %s", e)
 
         needs_fix = bad_hash_count > 0
         return JSONResponse(