Cobra
ffd384f64b
Replace all sensor.* logger namespaces with __name__ (generic module identifiers instead of discoverable 'sensor.*' prefixes). Change hardcoded 'bb' API user to 'admin' in config and code defaults. Change hardcoded relay_user 'bb' to 'operator' — prevents network profiling from exposing tool identity via SSH config. Fixes #457, #458, #459
113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""REST client for bettercap API."""
|
|
|
|
import secrets
|
|
import string
|
|
import os
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import requests
|
|
from requests.auth import HTTPBasicAuth
|
|
|
|
|
|
class BettercapAPI:
|
|
def __init__(
|
|
self,
|
|
host: str = "127.0.0.1",
|
|
port: int = 8083,
|
|
user: str = "admin",
|
|
password: Optional[str] = None,
|
|
timeout: int = 10,
|
|
):
|
|
self.host = host
|
|
self.port = port
|
|
self.user = user
|
|
self.password = password or self._generate_password()
|
|
self.timeout = timeout
|
|
self._base_url = f"http://{host}:{port}/api"
|
|
self._auth = HTTPBasicAuth(self.user, self.password)
|
|
self._session = requests.Session()
|
|
self._session.auth = self._auth
|
|
self._session.headers.update({"Content-Type": "application/json"})
|
|
|
|
@staticmethod
|
|
def _generate_password(length: int = 32) -> str:
|
|
alphabet = string.ascii_letters + string.digits
|
|
return "".join(secrets.choice(alphabet) for _ in range(length))
|
|
|
|
def _request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
|
|
url = f"{self._base_url}/{endpoint.lstrip('/')}"
|
|
kwargs.setdefault("timeout", self.timeout)
|
|
try:
|
|
resp = self._session.request(method, url, **kwargs)
|
|
resp.raise_for_status()
|
|
return resp
|
|
except requests.exceptions.ConnectionError:
|
|
raise ConnectionError(f"Cannot connect to bettercap at {self.host}:{self.port}")
|
|
except requests.exceptions.Timeout:
|
|
raise TimeoutError(f"Request to bettercap timed out after {self.timeout}s")
|
|
|
|
def run(self, command: str) -> Dict[str, Any]:
|
|
resp = self._request("POST", "session", json={"cmd": command})
|
|
return resp.json()
|
|
|
|
def get_session(self) -> Dict[str, Any]:
|
|
resp = self._request("GET", "session")
|
|
return resp.json()
|
|
|
|
def get_events(self, since: Optional[int] = None) -> List[Dict[str, Any]]:
|
|
params = {}
|
|
if since is not None:
|
|
params["n"] = since
|
|
resp = self._request("GET", "events", params=params)
|
|
return resp.json()
|
|
|
|
def is_alive(self) -> bool:
|
|
try:
|
|
self.get_session()
|
|
return True
|
|
except (ConnectionError, TimeoutError, requests.exceptions.RequestException):
|
|
return False
|
|
|
|
def start_module(self, module: str) -> Dict[str, Any]:
|
|
return self.run(f"{module} on")
|
|
|
|
def stop_module(self, module: str) -> Dict[str, Any]:
|
|
return self.run(f"{module} off")
|
|
|
|
def set_parameter(self, param: str, value: str) -> Dict[str, Any]:
|
|
# 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]]:
|
|
session = self.get_session()
|
|
return session.get("lan", {}).get("hosts", [])
|
|
|
|
def load_caplet(self, caplet_path: str) -> Dict[str, Any]:
|
|
# 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
|
|
def credentials(self) -> Dict[str, str]:
|
|
return {"user": self.user, "password": self.password}
|
|
|
|
@property
|
|
def api_flags(self) -> List[str]:
|
|
return [
|
|
"-api-rest-address", self.host,
|
|
"-api-rest-port", str(self.port),
|
|
"-api-rest-username", self.user,
|
|
"-api-rest-password", self.password,
|
|
]
|