Expand test coverage (35%→90%) and fix data-flow security issues

.github/workflows/ci.yml

@@ -29,4 +29,4 @@ jobs:
           pip install -r requirements-dev.txt
 
       - name: Run tests
-        run: pytest -q
+        run: pytest -q --cov=. --cov-report=term-missing --cov-fail-under=80

README.md

@@ -152,6 +152,8 @@ Environment variables (optional):
 | `FC_INVENTORY_PORT` | `5000` | Local web UI port |
 | `FC_INVENTORY_PASSWORD` | _(unset)_ | Password for headless `collect` mode (avoids putting it on the command line) |
 | `FC_INVENTORY_DISABLE_UPDATE_CHECK` | _(unset)_ | Set to any value to disable the "new version available" check in the web UI |
+| `FC_INVENTORY_VERIFY_SSL` | _(unset)_ | Set to `1`/`true` to verify the VRM's TLS certificate (off by default for self-signed certs) |
+| `FC_INVENTORY_CA_BUNDLE` | _(unset)_ | Path to a CA bundle/cert used to verify the VRM (enables verification against your own trust chain) |
 
 Example:
 ```cmd

app.py

@@ -13,6 +13,7 @@
 import threading
 import webbrowser
 from datetime import datetime
+from urllib.parse import urlparse
 
 from flask import Flask, render_template, request, jsonify, send_file
 
@@ -58,6 +59,26 @@
 }
 
 
+@app.before_request
+def _reject_cross_origin():
+    """Lightweight CSRF / drive-by protection for state-changing requests.
+
+    The tool has no login, so a malicious web page could otherwise POST to the
+    local server (CSRF). Browsers always attach an ``Origin`` header on
+    cross-origin requests; reject those whose origin host does not match the
+    request host. Non-browser clients (curl, the headless CLI) send no Origin
+    and are unaffected.
+    """
+    if request.method in ("GET", "HEAD", "OPTIONS"):
+        return None
+    origin = request.headers.get("Origin")
+    if not origin:
+        return None
+    if urlparse(origin).netloc != request.host:
+        return jsonify({"error": "Cross-origin request rejected."}), 403
+    return None
+
+
 def _run_collection(collector):
     """Background thread: collect data and build Excel file."""
     try:
@@ -104,13 +125,21 @@ def start_collection():
     # Parse request
     body = request.get_json(silent=True) or {}
     host = body.get("host", "").strip()
-    port = int(body.get("port", 7443))
     username = body.get("username", "").strip()
     password = body.get("password", "")
 
     if not host or not username or not password:
         return jsonify({"error": "Host, username, and password are required."}), 400
 
+    # Validate the port: reject non-numeric or out-of-range values up front
+    # instead of failing deep inside the request handler.
+    try:
+        port = int(body.get("port", 7443))
+    except (TypeError, ValueError):
+        return jsonify({"error": "Port must be a number."}), 400
+    if not 1 <= port <= 65535:
+        return jsonify({"error": "Port must be between 1 and 65535."}), 400
+
     # Clean up old output file
     if current_job["output_file"] and os.path.exists(current_job["output_file"]):
         try:

excel_builder.py

@@ -19,6 +19,23 @@
     "vHost", "vCluster", "vDatastore", "vSwitch",
 ]
 
+# Leading characters that spreadsheet apps may interpret as a formula.
+# Inventory values come from FusionCompute (a VM name/description could be
+# attacker-controlled), so neutralize them to prevent CSV/Excel injection.
+_FORMULA_TRIGGERS = ("=", "+", "-", "@")
+
+
+def _sanitize_cell(value):
+    """Defang spreadsheet-formula injection in string cell values.
+
+    A string beginning with '=', '+', '-' or '@' is prefixed with a single
+    quote so Excel/Sheets render it as literal text instead of evaluating it.
+    Non-string values are returned unchanged.
+    """
+    if isinstance(value, str) and value[:1] in _FORMULA_TRIGGERS:
+        return "'" + value
+    return value
+
 
 def build_excel(data, output_path):
     """
@@ -52,9 +69,9 @@ def build_excel(data, output_path):
                     headers.append(key)
         ws.append(headers)
 
-        # Write data rows
+        # Write data rows (sanitized against formula injection)
         for row in rows:
-            ws.append([row.get(h, "") for h in headers])
+            ws.append([_sanitize_cell(row.get(h, "")) for h in headers])
 
         # Apply header styling
         _style_header(ws)

fc_client.py

@@ -5,19 +5,36 @@
 
 import hashlib
 import logging
+import os
 import requests
 import urllib3
 
 logger = logging.getLogger(__name__)
 
-# Suppress SSL warnings for self-signed certificates
-urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+def _resolve_verify(verify):
+    """Decide the requests TLS-verify setting.
+
+    Precedence: explicit argument > FC_INVENTORY_CA_BUNDLE (path to a CA file)
+    > FC_INVENTORY_VERIFY_SSL (truthy enables verification) > default False.
+
+    FusionCompute VRM commonly uses self-signed certs, so verification is OFF
+    by default for backward compatibility, but operators can opt in to a
+    proper trust chain to defend against man-in-the-middle attacks.
+    """
+    if verify is not None:
+        return verify
+    ca_bundle = os.environ.get("FC_INVENTORY_CA_BUNDLE")
+    if ca_bundle:
+        return ca_bundle
+    flag = os.environ.get("FC_INVENTORY_VERIFY_SSL", "").strip().lower()
+    return flag in ("1", "true", "yes", "on")
 
 
 class FCClient:
     """Client for FusionCompute VRM REST API."""
 
-    def __init__(self, host, username, password, port=7443):
+    def __init__(self, host, username, password, port=7443, verify=None):
         # Strip protocol prefix if user entered it (e.g. https://10.0.0.1)
         host = host.replace("https://", "").replace("http://", "").strip("/")
         self.host = host
@@ -26,7 +43,10 @@ def __init__(self, host, username, password, port=7443):
         self.password = password
         self.base_url = f"https://{host}:{port}"
         self.session = requests.Session()
-        self.session.verify = False
+        self.session.verify = _resolve_verify(verify)
+        # Only suppress the insecure-request warning when verification is off.
+        if not self.session.verify:
+            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
         self.session.headers.update({
             "Content-Type": "application/json; charset=UTF-8",
             "Accept": "application/json;version=v1.0;charset=UTF-8",
@@ -111,17 +131,21 @@ def login(self):
                             )
 
                         logger.debug(f"  -> HTTP {resp.status_code}")
-                        try:
-                            logger.debug(f"  -> Body: {resp.text[:300]}")
-                        except Exception:
-                            pass
 
                         if resp.status_code == 200:
                             self.port = port
                             self.base_url = f"https://{self.host}:{port}/service"
                             logger.info(f"Login OK! version={ver}, base_url={self.base_url}")
                             return self._extract_token(resp, label)
 
+                        # Log the body only on failure: a successful login
+                        # response carries the session token, which must not
+                        # be written to the log file.
+                        try:
+                            logger.debug(f"  -> Body: {resp.text[:300]}")
+                        except Exception:
+                            pass
+
                         body = resp.text[:200]
                         logger.warning(f"  -> Failed ({resp.status_code}): {body}")
                         last_error = f"HTTP {resp.status_code}: {body}"
@@ -206,21 +230,39 @@ def _get(self, path, params=None):
         resp.raise_for_status()
         return resp.json()
 
+    @staticmethod
+    def _extract_list(data, *keys):
+        """Extract a list from an API response that may be a dict or bare list.
+
+        FusionCompute responses vary by version: some endpoints return a
+        wrapper dict ({"vms": [...]}), others a top-level list. Tries each
+        candidate key in order on a dict, falls back to the response itself
+        when it is already a list, and always returns a list.
+        """
+        if isinstance(data, list):
+            return data
+        if isinstance(data, dict):
+            for key in keys:
+                value = data.get(key)
+                if value is not None:
+                    return value
+        return []
+
     def _get_all(self, path, result_key):
         """Fetch all pages of a paginated list endpoint."""
         items = []
         offset = 0
         limit = 100
         while True:
             data = self._get(path, params={"offset": offset, "limit": limit})
-            # Try expected key, fallback to common alternatives
+            # If top-level is a list, use it directly (no pagination metadata).
+            if isinstance(data, list):
+                return data
+            # Try expected key, fallback to common alternatives.
             batch = data.get(result_key)
             if batch is None:
                 batch = data.get("items", data.get("result", []))
             if not batch:
-                # If top-level is a list, use it directly
-                if isinstance(data, list):
-                    return data
                 break
             items.extend(batch)
             total = data.get("total", len(items))
@@ -234,7 +276,7 @@ def _get_all(self, path, result_key):
     def get_sites(self):
         """Get all sites."""
         data = self._get("/sites")
-        return data.get("sites", [])
+        return self._extract_list(data, "sites")
 
     # ── Cluster ──────────────────────────────────────────
 
@@ -270,18 +312,18 @@ def get_vm_detail(self, vm_uri):
     def get_vm_nics(self, vm_uri):
         """Get network interfaces of a VM."""
         data = self._get(f"{vm_uri}/nics")
-        return data.get("nics", data.get("items", []))
+        return self._extract_list(data, "nics", "items")
 
     def get_vm_disks(self, vm_uri):
         """Get disk volumes of a VM."""
         # Try /volumes first, fallback to /disks
         try:
             data = self._get(f"{vm_uri}/volumes")
-            return data.get("volumes", data.get("items", []))
+            return self._extract_list(data, "volumes", "items")
         except Exception:
             try:
                 data = self._get(f"{vm_uri}/disks")
-                return data.get("disks", data.get("items", []))
+                return self._extract_list(data, "disks", "items")
             except Exception:
                 return []
 
@@ -297,28 +339,22 @@ def get_dvswitches(self, site_uri):
         """Get all distributed virtual switches."""
         data = self._get(f"{site_uri}/dvswitchs")
         logger.debug(f"DVSwitch response keys: {list(data.keys()) if isinstance(data, dict) else 'not a dict'}")
-        result = data.get("dvswitchs", data.get("dvSwitchs", data.get("items", [])))
-        if not result and isinstance(data, list):
-            result = data
+        result = self._extract_list(data, "dvswitchs", "dvSwitchs", "items")
         logger.info(f"Found {len(result)} DVSwitches")
         return result
 
     def get_portgroups(self, dvswitch_uri):
         """Get port groups of a DVSwitch. dvswitch_uri = full URI from dvswitch list."""
         data = self._get(f"{dvswitch_uri}/portgroups")
-        result = data.get("portgroups", data.get("portGroups", data.get("items", [])))
-        if not result and isinstance(data, list):
-            result = data
+        result = self._extract_list(data, "portgroups", "portGroups", "items")
         logger.info(f"Found {len(result)} port groups under {dvswitch_uri}")
         return result
 
     def get_site_portgroups(self, site_uri):
         """Get all port groups in a site (fallback when DVSwitch listing is empty)."""
         try:
             data = self._get(f"{site_uri}/portgroups")
-            result = data.get("portgroups", data.get("portGroups", data.get("items", [])))
-            if not result and isinstance(data, list):
-                result = data
+            result = self._extract_list(data, "portgroups", "portGroups", "items")
             logger.info(f"Found {len(result)} port groups at site level")
             return result
         except Exception as e:

requirements-dev.txt

@@ -1,2 +1,3 @@
 -r requirements.txt
 pytest>=8,<9
+pytest-cov>=5,<6