res/strategies.py
Rama master · réplica de lectura del repositorio original.
#!/usr/bin/env python3
import requests
import argparse
import json
def check_response(response):
"""
Check API response and handle errors.
Two error cases:
1. Status code is not 200 -> exit with error
2. Response contains {"error": "xxx"} -> exit with error
"""
if response.status_code != 200:
print(f"Error: HTTP {response.status_code}: {response.text}")
exit(1)
# Check for {"error": "xxx"} in response
if response.text and response.text.strip():
try:
json_data = response.json()
if isinstance(json_data, dict) and "error" in json_data:
print(f"Error: {json_data['error']}")
exit(1)
return json_data
except ValueError:
return response.text
return None
def headers_with(token):
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
# ---------- Strategies APIs ----------
def list_strategies(url, token):
"""List all strategies"""
headers = headers_with(token)
r = requests.get(f"{url}/api/strategies", headers=headers)
return check_response(r)
def get_strategy_by_guid(url, token, guid):
"""Get strategy by GUID"""
headers = headers_with(token)
r = requests.get(f"{url}/api/strategies/{guid}", headers=headers)
return check_response(r)
def get_strategy_by_name(url, token, name):
"""Get strategy by name"""
strategies = list_strategies(url, token)
if not strategies:
return None
for s in strategies:
if str(s.get("name")) == name:
return s
return None
def enable_strategy(url, token, name):
"""Enable a strategy"""
headers = headers_with(token)
strategy = get_strategy_by_name(url, token, name)
if not strategy:
print(f"Error: Strategy '{name}' not found")
exit(1)
guid = strategy.get("guid")
r = requests.put(f"{url}/api/strategies/{guid}/status", headers=headers, json=True)
check_response(r)
return "Success"
def disable_strategy(url, token, name):
"""Disable a strategy"""
headers = headers_with(token)
strategy = get_strategy_by_name(url, token, name)
if not strategy:
print(f"Error: Strategy '{name}' not found")
exit(1)
guid = strategy.get("guid")
r = requests.put(f"{url}/api/strategies/{guid}/status", headers=headers, json=False)
check_response(r)
return "Success"
def get_device_guid_by_id(url, token, device_id):
"""Get device GUID by device ID (exact match)"""
headers = headers_with(token)
params = {"id": device_id, "pageSize": 50}
r = requests.get(f"{url}/api/devices", headers=headers, params=params)
res = check_response(r)
if not res:
return None
devices_data = res.get("data", []) if isinstance(res, dict) else res
for d in devices_data:
if d.get("id") == device_id:
return d.get("guid")
return None
def get_user_guid_by_name(url, token, name):
"""Get user GUID by exact name match"""
headers = headers_with(token)
params = {"name": name, "pageSize": 50}
r = requests.get(f"{url}/api/users", headers=headers, params=params)
res = check_response(r)
if not res:
return None
users_data = res.get("data", []) if isinstance(res, dict) else res
for u in users_data:
if u.get("name") == name:
return u.get("guid")
return None
def get_device_group_guid_by_name(url, token, name):
"""Get device group GUID by exact name match"""
headers = headers_with(token)
params = {"pageSize": 50, "name": name}
r = requests.get(f"{url}/api/device-groups", headers=headers, params=params)
res = check_response(r)
if not res:
return None
groups_data = res.get("data", []) if isinstance(res, dict) else res
for g in groups_data:
if g.get("name") == name:
return g.get("guid")
return None
def assign_strategy(url, token, strategy_name, peers=None, users=None, device_groups=None):
"""
Assign strategy to peers, users, or device groups
Args:
strategy_name: Name of the strategy (or None to unassign)
peers: List of device IDs or GUIDs
users: List of user names or GUIDs
device_groups: List of device group names or GUIDs
"""
headers = headers_with(token)
# Get strategy GUID if strategy_name is provided
strategy_guid = None
if strategy_name:
strategy = get_strategy_by_name(url, token, strategy_name)
if not strategy:
print(f"Error: Strategy '{strategy_name}' not found")
exit(1)
strategy_guid = strategy.get("guid")
# Convert device IDs to GUIDs
peer_guids = []
if peers:
for peer in peers:
# Check if it's already a GUID format
if len(peer) == 36 and peer.count('-') == 4:
peer_guids.append(peer)
else:
# Treat as device ID, look it up
guid = get_device_guid_by_id(url, token, peer)
if not guid:
print(f"Error: Device '{peer}' not found")
exit(1)
peer_guids.append(guid)
# Convert user names to GUIDs
user_guids = []
if users:
for user in users:
# Check if it's already a GUID format
if len(user) == 36 and user.count('-') == 4:
user_guids.append(user)
else:
# Treat as username, look it up
guid = get_user_guid_by_name(url, token, user)
if not guid:
print(f"Error: User '{user}' not found")
exit(1)
user_guids.append(guid)
# Convert device group names to GUIDs
device_group_guids = []
if device_groups:
for dg in device_groups:
# Check if it's already a GUID format
if len(dg) == 36 and dg.count('-') == 4:
device_group_guids.append(dg)
else:
# Treat as device group name, look it up
guid = get_device_group_guid_by_name(url, token, dg)
if not guid:
print(f"Error: Device group '{dg}' not found")
exit(1)
device_group_guids.append(guid)
# Build payload
payload = {}
if strategy_guid:
payload["strategy"] = strategy_guid
payload["peers"] = peer_guids
payload["users"] = user_guids
payload["groups"] = device_group_guids
r = requests.post(f"{url}/api/strategies/assign", headers=headers, json=payload)
check_response(r)
def main():
parser = argparse.ArgumentParser(description="Strategy manager")
parser.add_argument("command", choices=[
"list", "view", "enable", "disable", "assign", "unassign"
])
parser.add_argument("--url", required=True, help="Server URL")
parser.add_argument("--token", required=True, help="API token")
parser.add_argument("--name", help="Strategy name (for view/enable/disable/assign commands)")
parser.add_argument("--guid", help="Strategy GUID (for view command, alternative to --name)")
# For assign/unassign commands
parser.add_argument("--peers", help="Comma separated device IDs or GUIDs (requires Device Permission:r)")
parser.add_argument("--users", help="Comma separated user names or GUIDs (requires User Permission:r)")
parser.add_argument("--device-groups", help="Comma separated device group names or GUIDs (requires Device Group Permission:r)")
args = parser.parse_args()
while args.url.endswith("/"): args.url = args.url[:-1]
if args.command == "list":
res = list_strategies(args.url, args.token)
print(json.dumps(res, indent=2))
elif args.command == "view":
if args.guid:
res = get_strategy_by_guid(args.url, args.token, args.guid)
print(json.dumps(res, indent=2))
elif args.name:
strategy = get_strategy_by_name(args.url, args.token, args.name)
if not strategy:
print(f"Error: Strategy '{args.name}' not found")
exit(1)
# Get full details by GUID
guid = strategy.get("guid")
res = get_strategy_by_guid(args.url, args.token, guid)
print(json.dumps(res, indent=2))
else:
print("Error: --name or --guid is required for view command")
exit(1)
elif args.command == "enable":
if not args.name:
print("Error: --name is required")
exit(1)
print(enable_strategy(args.url, args.token, args.name))
elif args.command == "disable":
if not args.name:
print("Error: --name is required")
exit(1)
print(disable_strategy(args.url, args.token, args.name))
elif args.command == "assign":
if not args.name:
print("Error: --name is required")
exit(1)
if not args.peers and not args.users and not args.device_groups:
print("Error: at least one of --peers, --users, or --device-groups is required")
exit(1)
peers = [x.strip() for x in args.peers.split(",") if x.strip()] if args.peers else None
users = [x.strip() for x in args.users.split(",") if x.strip()] if args.users else None
device_groups = [x.strip() for x in args.device_groups.split(",") if x.strip()] if args.device_groups else None
assign_strategy(args.url, args.token, args.name, peers=peers, users=users, device_groups=device_groups)
count = (len(peers) if peers else 0) + (len(users) if users else 0) + (len(device_groups) if device_groups else 0)
print(f"Success: Assigned strategy '{args.name}' to {count} target(s)")
elif args.command == "unassign":
if not args.peers and not args.users and not args.device_groups:
print("Error: at least one of --peers, --users, or --device-groups is required")
exit(1)
peers = [x.strip() for x in args.peers.split(",") if x.strip()] if args.peers else None
users = [x.strip() for x in args.users.split(",") if x.strip()] if args.users else None
device_groups = [x.strip() for x in args.device_groups.split(",") if x.strip()] if args.device_groups else None
assign_strategy(args.url, args.token, None, peers=peers, users=users, device_groups=device_groups)
count = (len(peers) if peers else 0) + (len(users) if users else 0) + (len(device_groups) if device_groups else 0)
print(f"Success: Unassigned strategy from {count} target(s)")
if __name__ == "__main__":
main()