Files
Samsara/samsara/analyzers/assessment.py

45 lines
3.2 KiB
Python

def assess(data):
inventory = data["inventory"]
findings = []
def add(category, classification, evidence, severity, points):
findings.append(dict(category=category, classification=classification, evidence=evidence,
severity=severity, points=points))
for row in inventory.get("roles", []):
add("Installed role", "Potential dependency", row["name"], "medium", 10)
linux_apps = ('nginx', 'apache2', 'httpd', 'mysql', 'mysqld', 'mariadb', 'postgresql',
'redis', 'docker', 'containerd', 'smbd', 'smb', 'nfs-server', 'named', 'haproxy', 'tomcat')
unclassified = []
for row in inventory.get("services", []):
if row.get("state") == "Running":
if data['server'].get('platform') == 'linux' and not any(
row['name'].split('@')[0].split('.')[0] == app or row['name'].startswith(app + '-')
for app in linux_apps
):
unclassified.append(row['name'])
continue
add("Active service", "Potential dependency", f"{row['name']} running as {row.get('account', 'unknown')}", "medium", 10)
if unclassified:
add("Unclassified services", "Unknown or insufficient evidence",
f"{len(unclassified)} running Linux services lack an application rule. Administrator review required; examples: " + ', '.join(unclassified[:8]), "low", 5)
for row in inventory.get("network", []):
if row.get("state") == "Established":
add("Network activity", "Confirmed observed activity", f"{row['local_address']}:{row['local_port']} ↔ {row['remote_address']}:{row['remote_port']}; direction unverified", "high", 20)
elif row.get("state") == "Listen":
add("Listening port", "Potential dependency", f"TCP {row['local_port']} ({row.get('process', 'unknown')})", "medium", 5)
for row in inventory.get("tasks", []):
if row.get("enabled"):
add("Scheduled workload", "Potential dependency", f"{row['name']}; account {row.get('account', 'unknown')}", "medium", 10)
for row in inventory.get("storage", []):
if row.get("type") in ("SMB Share", "NFS Export", "Samba Share"):
add("Shared storage", "Potential dependency", f"{row['name']}: {row['path']}", "high", 15)
for row in inventory.get("storage", []):
if row.get("type") == "Remote Mount":
add("Remote storage consumer", "Potential dependency", f"{row['name']} mounted at {row['path']}; access and use unverified", "medium", 10)
for name, check in data["checks"].items():
if check["status"] != "complete":
add("Missing evidence", "Unknown or insufficient evidence", f"{name}: {check['message']}", "medium", 10)
add("Observation limits", "Unknown or insufficient evidence", "Point-in-time inventory cannot establish infrequent, offline, historical or external dependencies. No active connections does not prove a server is unused.", "low", 5)
score = min(100, sum(f["points"] for f in findings))
level = "critical" if score >= 90 else "high" if score >= 60 else "medium" if score >= 30 else "low" if score else "informational"
return {"score": score, "level": level, "findings": findings}