Fix #207: Prevent command injection in bettercap_api.py
- set_parameter() now validates param name with regex (alphanumeric, dots, underscores) - set_parameter() quotes value with repr() to prevent argument injection - load_caplet() validates path exists, is absolute, and ends with .cap/.caplet - load_caplet() quotes caplet path for safe interpolation into bettercap command
This commit is contained in:
+16
-2
@@ -3,6 +3,7 @@
|
|||||||
|
|
||||||
import secrets
|
import secrets
|
||||||
import string
|
import string
|
||||||
|
import os
|
||||||
from typing import Any, Dict, List, Optional
|
from typing import Any, Dict, List, Optional
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
@@ -75,14 +76,27 @@ class BettercapAPI:
|
|||||||
return self.run(f"{module} off")
|
return self.run(f"{module} off")
|
||||||
|
|
||||||
def set_parameter(self, param: str, value: str) -> Dict[str, Any]:
|
def set_parameter(self, param: str, value: str) -> Dict[str, Any]:
|
||||||
return self.run(f"set {param} {value}")
|
# Validate param to prevent injection — allow alphanumeric, dots, underscores
|
||||||
|
if not re.match(r'^[a-zA-Z0-9._-]+$', param):
|
||||||
|
raise ValueError(f"Invalid parameter name: {param}")
|
||||||
|
# Quote value to prevent injection
|
||||||
|
quoted_value = repr(value)
|
||||||
|
return self.run(f"set {param} {quoted_value}")
|
||||||
|
|
||||||
def get_hosts(self) -> List[Dict[str, Any]]:
|
def get_hosts(self) -> List[Dict[str, Any]]:
|
||||||
session = self.get_session()
|
session = self.get_session()
|
||||||
return session.get("lan", {}).get("hosts", [])
|
return session.get("lan", {}).get("hosts", [])
|
||||||
|
|
||||||
def load_caplet(self, caplet_path: str) -> Dict[str, Any]:
|
def load_caplet(self, caplet_path: str) -> Dict[str, Any]:
|
||||||
return self.run(f"caplet.load {caplet_path}")
|
# Validate caplet path — must be absolute, must exist, prevent traversal
|
||||||
|
caplet_path = os.path.abspath(caplet_path)
|
||||||
|
if not os.path.isfile(caplet_path):
|
||||||
|
raise FileNotFoundError(f"Caplet not found: {caplet_path}")
|
||||||
|
if not caplet_path.endswith(('.cap', '.caplet')):
|
||||||
|
raise ValueError(f"Invalid caplet extension: {caplet_path}")
|
||||||
|
# Quote path for safety
|
||||||
|
quoted_path = repr(caplet_path)
|
||||||
|
return self.run(f"caplet.load {quoted_path}")
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def credentials(self) -> Dict[str, str]:
|
def credentials(self) -> Dict[str, str]:
|
||||||
|
|||||||
Reference in New Issue
Block a user