res/audits.py
Rama master · réplica de lectura del repositorio original.
#!/usr/bin/env python3
import requests
import argparse
import json
from datetime import datetime, timedelta, timezone
def format_timestamp(timestamp):
"""Convert Unix timestamp to readable local datetime"""
if timestamp is None:
return None
try:
# Convert to local time
local_dt = datetime.fromtimestamp(timestamp)
return local_dt.strftime("%Y-%m-%d %H:%M:%S")
except (ValueError, TypeError):
return timestamp
def parse_local_time_to_utc_string(time_str):
"""Parse local time string to UTC time string for API filtering"""
try:
# Parse the local time string
local_dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S.%f")
# Make the datetime object timezone-aware using system's local timezone
local_dt = local_dt.replace(tzinfo=datetime.now().astimezone().tzinfo)
utc_dt = local_dt.astimezone(timezone.utc)
return utc_dt.strftime("%Y-%m-%d %H:%M:%S.000")
except ValueError:
try:
# Try without microseconds
local_dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
# Make the datetime object timezone-aware using system's local timezone
local_dt = local_dt.replace(tzinfo=datetime.now().astimezone().tzinfo)
utc_dt = local_dt.astimezone(timezone.utc)
return utc_dt.strftime("%Y-%m-%d %H:%M:%S.000")
except ValueError:
return None
def get_connection_type_name(conn_type):
"""Convert connection type number to readable name"""
type_map = {
0: "Remote Desktop",
1: "File Transfer",
2: "Port Transfer",
3: "View Camera",
4: "Terminal"
}
return type_map.get(conn_type, f"Unknown ({conn_type})")
def get_console_type_name(console_type):
"""Convert console audit type number to readable name"""
type_map = {
0: "Group Management",
1: "User Management",
2: "Device Management",
3: "Address Book Management"
}
return type_map.get(console_type, f"Unknown ({console_type})")
def get_console_operation_name(operation_code):
"""Convert console operation code to readable name"""
operation_map = {
0: "User Login",
1: "Add Group",
2: "Add User",
3: "Add Device",
4: "Delete Groups",
5: "Disconnect Device",
6: "Enable Users",
7: "Disable Users",
8: "Enable Devices",
9: "Disable Devices",
10: "Update Group",
11: "Update User",
12: "Update Device",
13: "Delete User",
14: "Delete Device",
15: "Add Address Book",
16: "Delete Address Book",
17: "Change Address Book Name",
18: "Delete Devices in the Address Book Recycle Bin",
19: "Empty Address Book Recycle Bin",
20: "Add Address Book Permission",
21: "Delete Address Book Permission",
22: "Update Address Book Permission"
}
return operation_map.get(operation_code, f"Unknown ({operation_code})")
def get_alarm_type_name(alarm_type):
"""Convert alarm type number to readable name"""
type_map = {
0: "Access attempt outside the IP whitelist",
1: "Over 30 consecutive access attempts",
2: "Multiple access attempts within one minute",
3: "Over 30 consecutive login attempts",
4: "Multiple login attempts within one minute",
5: "Multiple login attempts within one hour"
}
return type_map.get(alarm_type, f"Unknown ({alarm_type})")
def enhance_audit_data(data, audit_type):
"""Enhance audit data with readable formats"""
if not data:
return data
enhanced_data = []
for item in data:
enhanced_item = item.copy()
# Convert timestamps - replace original values
if 'created_at' in enhanced_item:
enhanced_item['created_at'] = format_timestamp(enhanced_item['created_at'])
if 'end_time' in enhanced_item:
enhanced_item['end_time'] = format_timestamp(enhanced_item['end_time'])
# Add type-specific enhancements - replace original values
if audit_type == 'conn':
if 'conn_type' in enhanced_item:
enhanced_item['conn_type'] = get_connection_type_name(enhanced_item['conn_type'])
else:
enhanced_item['conn_type'] = "Not Logged In"
elif audit_type == 'console':
if 'typ' in enhanced_item:
# Replace typ field with type and convert to readable name
enhanced_item['type'] = get_console_type_name(enhanced_item['typ'])
del enhanced_item['typ']
if 'iop' in enhanced_item:
# Replace iop field with operation and convert to readable name
enhanced_item['operation'] = get_console_operation_name(enhanced_item['iop'])
del enhanced_item['iop']
elif audit_type == 'alarm' and 'typ' in enhanced_item:
# Replace typ field with type and convert to readable name
enhanced_item['type'] = get_alarm_type_name(enhanced_item['typ'])
del enhanced_item['typ']
enhanced_data.append(enhanced_item)
return enhanced_data
def check_response(response):
"""Check API response and return result"""
if response.status_code != 200:
print(f"Error: HTTP {response.status_code} - {response.text}")
exit(1)
try:
response_json = response.json()
if "error" in response_json:
print(f"Error: {response_json['error']}")
exit(1)
return response_json
except ValueError:
return response.text or "Success"
def view_audits_common(url, token, endpoint, filters=None, page_size=None, current=None,
created_at=None, days_ago=None, non_wildcard_fields=None):
"""Common function for viewing audits"""
headers = {"Authorization": f"Bearer {token}"}
# Set default page size and current page
if page_size is None:
page_size = 10
if current is None:
current = 1
params = {
"pageSize": page_size,
"current": current
}
# Add filter parameters if provided
if filters:
for key, value in filters.items():
if value is not None:
params[key] = value
# Handle time filters
if days_ago is not None:
# Calculate datetime from days ago
target_time = datetime.now() - timedelta(days=days_ago)
# Convert to UTC time string using system timezone
utc_timestamp = target_time.timestamp()
utc_dt = datetime.fromtimestamp(utc_timestamp, timezone.utc)
params["created_at"] = utc_dt.strftime("%Y-%m-%d %H:%M:%S.000")
elif created_at:
# Parse local time string and convert to UTC time string
utc_time_str = parse_local_time_to_utc_string(created_at)
if utc_time_str is not None:
params["created_at"] = utc_time_str
else:
# If parsing fails, pass the original value
params["created_at"] = created_at
# Apply wildcard patterns for string fields (excluding specific fields)
if non_wildcard_fields is None:
non_wildcard_fields = set()
# Always exclude these fields from wildcard treatment
non_wildcard_fields.update(["created_at", "pageSize", "current"])
string_params = {}
for k, v in params.items():
if isinstance(v, str) and k not in non_wildcard_fields:
if v != "-" and "%" not in v:
string_params[k] = "%" + v + "%"
else:
string_params[k] = v
else:
string_params[k] = v
response = requests.get(f"{url}/api/audits/{endpoint}", headers=headers, params=string_params)
response_json = check_response(response)
# Enhance the data with readable formats
data = enhance_audit_data(response_json.get("data", []), endpoint)
return {
"data": data,
"total": response_json.get("total", 0),
"current": current,
"pageSize": page_size
}
def view_conn_audits(url, token, remote=None, conn_type=None,
page_size=None, current=None, created_at=None, days_ago=None):
"""View connection audits"""
filters = {
"remote": remote,
"conn_type": conn_type
}
non_wildcard_fields = {"conn_type"}
return view_audits_common(
url, token, "conn", filters, page_size, current, created_at, days_ago, non_wildcard_fields
)
def view_file_audits(url, token, remote=None,
page_size=None, current=None, created_at=None, days_ago=None):
"""View file audits"""
filters = {
"remote": remote
}
non_wildcard_fields = set()
return view_audits_common(
url, token, "file", filters, page_size, current, created_at, days_ago, non_wildcard_fields
)
def view_alarm_audits(url, token, device=None,
page_size=None, current=None, created_at=None, days_ago=None):
"""View alarm audits"""
filters = {
"device": device
}
non_wildcard_fields = set()
return view_audits_common(
url, token, "alarm", filters, page_size, current, created_at, days_ago, non_wildcard_fields
)
def view_console_audits(url, token, operator=None,
page_size=None, current=None, created_at=None, days_ago=None):
"""View console audits"""
filters = {
"operator": operator
}
non_wildcard_fields = set()
return view_audits_common(
url, token, "console", filters, page_size, current, created_at, days_ago, non_wildcard_fields
)
def main():
parser = argparse.ArgumentParser(description="Audits manager")
parser.add_argument(
"command",
choices=["view-conn", "view-file", "view-alarm", "view-console"],
help="Command to execute",
)
parser.add_argument("--url", required=True, help="URL of the API")
parser.add_argument("--token", required=True, help="Bearer token for authentication")
# Pagination parameters
parser.add_argument("--page-size", type=int, default=10, help="Number of records per page (default: 10)")
parser.add_argument("--current", type=int, default=1, help="Current page number (default: 1)")
# Time filtering parameters
parser.add_argument("--created-at", help="Filter by creation time in local time (format: 2025-09-16 14:15:57 or 2025-09-16 14:15:57.000)")
parser.add_argument("--days-ago", type=int, help="Filter by days ago (e.g., 7 for last 7 days)")
# Audit filters (simplified)
parser.add_argument("--remote", help="Remote peer ID filter (for conn/file audits)")
parser.add_argument("--device", help="Device ID filter (for alarm audits)")
parser.add_argument("--conn-type", type=int, help="Connection type filter (for conn audits only): 0=Remote Desktop, 1=File Transfer, 2=Port Transfer, 3=View Camera, 4=Terminal")
parser.add_argument("--operator", help="Operator filter (for console audits only)")
args = parser.parse_args()
# Remove trailing slashes from URL
while args.url.endswith("/"):
args.url = args.url[:-1]
if args.command == "view-conn":
# View connection audits
result = view_conn_audits(
args.url,
args.token,
args.remote,
args.conn_type,
args.page_size,
args.current,
args.created_at,
args.days_ago
)
print(json.dumps(result, indent=2))
elif args.command == "view-file":
# View file audits
result = view_file_audits(
args.url,
args.token,
args.remote,
args.page_size,
args.current,
args.created_at,
args.days_ago
)
print(json.dumps(result, indent=2))
elif args.command == "view-alarm":
# View alarm audits
result = view_alarm_audits(
args.url,
args.token,
args.device,
args.page_size,
args.current,
args.created_at,
args.days_ago
)
print(json.dumps(result, indent=2))
elif args.command == "view-console":
# View console audits
result = view_console_audits(
args.url,
args.token,
args.operator,
args.page_size,
args.current,
args.created_at,
args.days_ago
)
print(json.dumps(result, indent=2))
if __name__ == "__main__":
main()