通过API调用堡垒机的好处主要包括以下几点：
- 自动化和简化管理：API调用可以自动化堡垒机的配置和管理任务，减少手动操作的复杂性和出错的可能性。例如，通过API可以批量创建、修改或删除堡垒机实例，以及自动化处理与堡垒机相关的网络、安全组和路由规则等配置。
- 提高安全性：API调用可以集成到更广泛的安全策略中，确保所有操作都符合安全合规要求。例如，可以利用API对堡垒机的访问权限进行细粒度控制，确保只有授权的用户或系统能够执行特定的操作。
- 增强审计和监控能力：通过API可以更容易地集成堡垒机的审计日志和监控数据到企业的中央审计系统或安全信息和事件管理（SIEM）系统中，从而实现对运维行为的实时监控和事后审计。
- 提升运维效率：API调用可以简化运维流程，提高运维效率。例如，通过API可以实现快速的资源调配和问题响应，减少等待和处理时间。
- 支持全球化部署：对于需要在全球多个地区部署堡垒机的企业，API调用可以支持全球化的资源管理和运维，确保不同地区的资源能够统一管理和控制。
- 灵活性和可扩展性：API调用允许企业根据业务需求灵活地调整堡垒机的配置和策略，同时也可以方便地扩展新的功能和服务，以适应不断变化的业务需求。
- 减少直接暴露风险：通过API调用，可以减少对直接连接的需求，从而降低被攻击的风险。例如，可以配置API调用使得堡垒机在没有外网访问的条件下，进行安全的访问如云服务、数据库等。
- 支持多平台和协议：API调用通常支持多种操作系统平台和网络协议，使得不同环境下的资源都能够通过统一的方式进行管理和访问
删除jumpserver资产
import requests serverip = "http://10.12.2.132" def get_token(): url = serverip + '/api/v1/authentication/auth/' query_args = { "username": "admin", "password": "123456" } response = requests.post(url, data=query_args) return response.json().get('token') def get_all_assets(token): url = serverip + '/api/v1/assets/hosts/' headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} response = requests.get(url, headers=headers) return response.json() def delete_asset(asset_id, token): url = f"{serverip}/api/v1/assets/hosts/{asset_id}/" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} response = requests.delete(url, headers=headers) return response.status_code, response.text def delete_all_assets(): token = get_token() assets = get_all_assets(token) for asset in assets: asset_id = asset['id'] status_code, response_text = delete_asset(asset_id, token) print(f"Deleted asset {asset_id}: Status {status_code}, Response: {response_text}") if __name__ == "__main__": delete_all_assets()
删除/获取jumpserver的节点（nodes）
删除
import requests serverip = "http://10.12.2.132" def get_token(): url = serverip + '/api/v1/authentication/auth/' query_args = { "username": "admin", "password": "123456" } response = requests.post(url, data=query_args) return response.json().get('token') def get_all_nodes(token): url = serverip + '/api/v1/assets/nodes/' headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} response = requests.get(url, headers=headers) return response.json() def delete_node(node_id, token): url = f"{serverip}/api/v1/assets/nodes/{node_id}/" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} response = requests.delete(url, headers=headers) return response.status_code, response.text def delete_all_nodes(): token = get_token() nodes = get_all_nodes(token) for node in nodes: node_id = node['id'] status_code, response_text = delete_node(node_id, token) print(f"Deleted node {node_id}: Status {status_code}, Response: {response_text}") if __name__ == "__main__": delete_all_nodes()
获取
import requests serverip = "http://10.12.2.132" def get_token(): url = serverip + '/api/v1/authentication/auth/' query_args = { "username": "admin", "password": "123456" } response = requests.post(url, data=query_args) return response.json().get('token') token = get_token() # 请求头 headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } # 获取节点列表 response = requests.get(f"{serverip}/api/v1/assets/nodes/", headers=headers) if response.status_code == 200: nodes = response.json() for node in nodes: node_id = node.get('id') node_name = node.get('name') print(f"Node ID: {node_id}, Node Name: {node_name}") else: print("Failed to retrieve nodes:", response.text)
修改jumpserver登录用户密码
import requests import json # 配置 jumpserver_domain = "http://10.12.2.132" # 获取jumpserver认证token def get_token(): url = jumpserver_domain + '/api/v1/authentication/auth/' query_args = { "username": "admin", "password": "123456" } response = requests.post(url, data=query_args) return response.json().get('token') # 获取jumpserver所有用户ID def get_jm_userId(): api_token = get_token() # 请求头 headers = { "Authorization": f"Bearer {api_token}", "Content-Type": "application/json" } response = requests.get(f"{jumpserver_domain}/api/v1/users/users/", headers=headers) # 检查响应 if response.status_code == 200: users = response.json() # 用来保存除 Administrator 所有用户ID user_ids = [] for user in users: if user["name"] == "Administrator": continue user_ids.append(user['id']) return user_ids else: print("Failed to retrieve users:", response.text) # 设置用户密码 def set_all_jmUser_PassWd(user_id,new_password): api_token = get_token() headers = { 'Content-Type': 'application/json', 'Authorization': f"Bearer {api_token}" # 替换为您的实际 token } update_response = requests.patch( f"{jumpserver_domain}/api/v1/users/users/{user_id}/", headers=headers, json={"password": new_password} ) if update_response.status_code == 200: print(f"Password updated for user ID: {user_id}") else: print(f"Failed to update password for user ID: {user_id}", update_response.text) if __name__ == '__main__': user_id = get_jm_userId() for id_all in user_id: set_all_jmUser_PassWd(id_all, "AaBb123456")
文章评论