Security Audit & CI/CD Integration: Performed a security audit of the codebase, resolved all Medium and High severity issues, and integrated automated security scanning into the CI/CD pipeline.

Author: Khanz9664

.bandit.yaml

@@ -0,0 +1,19 @@
+# Bandit configuration file
+# Ref: https://bandit.readthedocs.io/en/latest/config.html
+
+# Exclude files from scanning
+exclude_dirs: ['tests', 'venv', '.venv', '__pycache__', '.git']
+
+# Explicitly skip certain tests for specific files
+# (Not supported in .bandit.yaml directly, but we can configure global skips)
+# we will use the --skips flag in CI, but here we define the basics
+
+# skips: ['B101'] # We could skip B101 globally, but it's better to exclude tests dir
+
+# Profiles
+profiles:
+    clinical_audit:
+        include:
+            - any_other_test_id_here
+        exclude:
+            - B101 # Assert used (common in tests, but we excluded tests dir already)

.github/workflows/pipeline.yml

@@ -23,6 +23,25 @@ jobs:
         run: |
           flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
 
+  security-audit:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python 3.12
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+      - name: Install Security Tools
+        run: |
+          pip install bandit safety
+      - name: Bandit Security Scan
+        run: |
+          # Scan excluding tests and venv, fail on Medium severity or higher
+          bandit -r . -x ./tests,./venv,./.venv -ll
+      - name: Safety Dependency Audit
+        run: |
+          safety check -r requirements.txt
+
   test:
     needs: lint
     runs-on: ubuntu-latest

api/main.py

@@ -205,4 +205,6 @@ def get_monitoring_status():
     }
 
 if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    host = os.getenv("HOST", "127.0.0.1")
+    port = int(os.getenv("PORT", 8000))
+    uvicorn.run(app, host=host, port=port)

app/main.py

@@ -52,7 +52,6 @@
 from src.monitoring.engine import MonitoringEngine
 from app.components.footer import render_footer
 import streamlit.components.v1 as components
-import subprocess
 import sys
 
 # Page Config
@@ -263,7 +262,7 @@ def user_input_features():
     st.info(f"Medical Grade AI | v{metadata.get('version', '1.0.0') if metadata else 'N/A'} Production")
     st.sidebar.markdown("---")
     st.sidebar.caption(f" **Clinical Engine**: v{metadata.get('version', '1.0.0') if metadata else 'N/A'}")
-    st.sidebar.caption(f" **Audit Hash**: {hashlib.md5(str(metadata).encode()).hexdigest()[:8].upper() if metadata else 'N/A'}")
+    st.sidebar.caption(f" **Audit Hash**: {hashlib.md5(str(metadata).encode(), usedforsecurity=False).hexdigest()[:8].upper() if metadata else 'N/A'}")
 
 # --- MAIN DASHBOARD ---
 # 1. Safety & Trust Header
@@ -407,7 +406,7 @@ def user_input_features():
     # Check for simulation state
     opt_full = st.session_state.get('last_opt_full', None)
     confidence = safety_engine.calculate_confidence(probability[0][1])
-    audit_hash = hashlib.md5(str(metadata).encode()).hexdigest()[:12].upper()
+    audit_hash = hashlib.md5(str(metadata).encode(), usedforsecurity=False).hexdigest()[:12].upper()
 
     radar_plot_path = None
     if opt_full:

logs/cardiosense.log

@@ -342,3 +342,11 @@ ValueError: [TypeError("'numpy.bool' object is not iterable"), TypeError('vars()
 2026-04-06 21:22:18,632 | INFO     | [API-ENGINE] | REQ [39751c2f-209d-4730-a0a4-0837150ae755] | GET /health | STATUS: 200 | DUR: 0.0020s
 2026-04-06 21:22:35,605 | INFO     | [API-ENGINE] | REQ [b1925b6f-7680-470c-89a4-ae78ce9c1f7c] | GET /monitoring/status | STATUS: 200 | DUR: 0.5281s
 2026-04-06 21:23:03,067 | INFO     | [API-ENGINE] | Shutting down Clinical Intelligence System...
+2026-04-11 11:24:58,739 | INFO     | [API-ENGINE] | Initializing Clinical Intelligence System...
+2026-04-11 11:24:58,777 | INFO     | [API-ENGINE] | Predictor model and preprocessor successfully loaded.
+2026-04-11 11:24:58,778 | INFO     | [API-ENGINE] | Clinical metadata loaded. Active Model Version: 2.4.0
+2026-04-11 11:25:05,575 | INFO     | [API-ENGINE] | REQ [fdab21b0-bef9-4eda-bd57-f187ff0b3441] | GET / | STATUS: 200 | DUR: 0.0041s
+2026-04-11 11:25:06,283 | INFO     | [API-ENGINE] | REQ [7c1ce280-1cbf-455b-8087-57a27827cf74] | GET /favicon.ico | STATUS: 404 | DUR: 0.0005s
+2026-04-11 11:25:11,432 | INFO     | [API-ENGINE] | REQ [0df41d96-6591-4aa3-8a81-9a70c28ada4c] | GET /docs | STATUS: 200 | DUR: 0.0005s
+2026-04-11 11:25:12,193 | INFO     | [API-ENGINE] | REQ [1e75a09c-be4d-4f92-9ef7-b69c598e5d46] | GET /openapi.json | STATUS: 200 | DUR: 0.0169s
+2026-04-11 11:25:27,862 | INFO     | [API-ENGINE] | Shutting down Clinical Intelligence System...

src/monitoring/engine.py

@@ -111,9 +111,9 @@ def run_drift_analysis(self, window_size: int = 500) -> Dict[str, Any]:
                 drift_report.save_html(html_path)
             elif hasattr(drift_report, 'save'):
                 drift_report.save(html_path)
-        except Exception:
-            # Silent fallback for environment-specific rendering limits
-            pass
+        except Exception as e:
+            # Fallback for environment-specific rendering limits
+            self._log(f"Report HTML save failed: {e}", "WARNING")
 
         # Extract Summary Metrics (Bypassing missing methods via Direct Attribute Access)
         drift_share = 0.0

src/monitoring/logger.py

@@ -69,8 +69,8 @@ def log_feedback(self, request_id: str, outcome: int):
     def get_recent_logs(self, limit: int = 1000) -> pd.DataFrame:
         """Retrieves recent logs as a pandas DataFrame for drift analysis."""
         conn = sqlite3.connect(self.db_path)
-        query = f"SELECT * FROM inference_logs ORDER BY timestamp DESC LIMIT {limit}"
-        df = pd.read_sql_query(query, conn)
+        query = "SELECT * FROM inference_logs ORDER BY timestamp DESC LIMIT ?"
+        df = pd.read_sql_query(query, conn, params=(int(limit),))
         conn.close()
 
         if df.empty:

src/recommendation/engine.py

@@ -113,7 +113,10 @@ def generate_prioritized_recommendations(self, data: pd.DataFrame, probability:
                             "rationale": f"SHAP analysis identifies {feat.upper()} as a top driver for this patient's risk profile.",
                             "type": "Data-Driven"
                         })
-            except Exception:
+            except Exception as e:
+                # In a clinical environment, we log the failure to generate a specific recommendation
+                # but allow the engine to proceed with other recommendations.
+                print(f"DEBUG: Recommendation generation failed for factor {feat}: {e}")
                 continue
 
         # Sort: High -> Moderate -> Low

src/utils/report_generator.py

@@ -15,7 +15,7 @@ def __init__(self, logo_path="app/assets/logo.png", audit_hash=None, version="2.
         self.version = version
         self.set_auto_page_break(auto=True, margin=15)
         self.report_date = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
-        self.report_id = hashlib.md5(self.report_date.encode()).hexdigest()[:8].upper()
+        self.report_id = hashlib.md5(self.report_date.encode(), usedforsecurity=False).hexdigest()[:8].upper()
         self.audit_hash = audit_hash or "N/A"
 
     def header(self):