import os import requests import json import uuid import argparse parser = argparse.ArgumentParser(description="Portainer helper - use env vars or pass credentials.") parser.add_argument("--base", "-b", default=os.getenv("PORTAINER_URL", "https://portainer.example.com"), help="Base URL for Portainer (ENV: PORTAINER_URL)") parser.add_argument("--user", "-u", default=os.getenv("PORTAINER_USER"), help="Portainer username (ENV: PORTAINER_USER)") parser.add_argument("--password", "-p", default=os.getenv("PORTAINER_PASS"), help="Portainer password (ENV: PORTAINER_PASS)") parser.add_argument("--endpoint-id", "-e", type=int, help="Endpoint ID to limit stack operations") parser.add_argument("--list-endpoints", action="store_true", help="List endpoints") parser.add_argument("--list-stacks", action="store_true", help="List stacks") parser.add_argument("--get-stack", metavar="NAME_OR_ID", help="Get stack by name or numeric id") parser.add_argument("--create-stack", action='store_true') parser.add_argument("--stack-id", "-s", type=str, help="Stack ID for operations") parser.add_argument("--token-only", action="store_true", help="Print auth token and exit") parser.add_argument("--timeout", type=int, default=10, help="Request timeout seconds") args = parser.parse_args() def get_portainer_token(base_url, username=None, password=None, timeout=10): """ Authenticate to Portainer and return a JWT token. Reads PORTAINER_USER / PORTAINER_PASS from environment if username/password are not provided. """ username = username or os.getenv("PORTAINER_USER") password = password or os.getenv("PORTAINER_PASS") if not username or not password: raise ValueError("Username and password must be provided (or set PORTAINER_USER / PORTAINER_PASS).") url = f"{base_url.rstrip('/')}/api/auth" resp = requests.post(url, json={"Username": username, "Password": password}, timeout=timeout) resp.raise_for_status() data = resp.json() token = data.get("jwt") or data.get("JWT") or data.get("token") if not token: raise ValueError(f"No token found in response: {data}") return token def api_get(base_url, path, token, timeout=10): """Example authenticated GET request to Portainer API.""" url = f"{base_url.rstrip('/')}{path}" headers = {"Authorization": f"Bearer {token}"} resp = requests.get(url, headers=headers, timeout=timeout) resp.raise_for_status() return resp.json() def api_post(base_url, path, token, data, timeout=20): """Example authenticated GET request to Portainer API.""" url = f"{base_url.rstrip('/')}{path}" headers = {"Authorization": f"Bearer {token}"} resp = requests.post(url, headers=headers, json=data, timeout=timeout) resp.raise_for_status() return resp.json() def get_stacks(base_url, token, endpoint_id=None, timeout=10): """ Return a list of stacks. If endpoint_id is provided, it will be added as a query param. """ path = "/api/stacks" if endpoint_id is not None: path += f"?endpointId={endpoint_id}" stacks = api_get(base_url, path, token, timeout=timeout) if stacks is None: return [] return stacks def get_stack(base_url, identifier, token, endpoint_id=None, timeout=10): """ Retrieve a single stack by numeric Id or by Name. Identifier may be an int (Id) or a string (Name). Raises ValueError if not found. """ stacks = get_stacks(base_url, token, endpoint_id=endpoint_id, timeout=timeout) # Normalize identifier ident_id = None try: ident_id = int(identifier) except (TypeError, ValueError): pass for s in stacks: # Many Portainer responses use 'Id' and 'Name' keys if ident_id is not None and s.get("Id") == ident_id: return s if str(s.get("Name")) == str(identifier): return s raise ValueError(f"Stack not found: {identifier}") def create_stack(base_url, token, endpoint_id=None, data={}, timeout=10): """ Return a list of stacks. If endpoint_id is provided, it will be added as a query param. """ path = "/api/stacks/create/standalone/repository" if endpoint_id is not None: path += f"?endpointId={endpoint_id}" try: stacks = api_post(base_url, path, token, data, timeout=timeout) except Exception as e: #print(f"Error creating stack: {e}") if "Conflict for url" in str(e): print("Stack with this name may already exist.") return [] if stacks is None: return [] return stacks if __name__ == "__main__": # Example usage: set PORTAINER_USER and PORTAINER_PASS in env, or pass literals below. base = os.getenv("PORTAINER_URL", "https://portainer.sectorq.eu") token = get_portainer_token(base,"admin","l4c1j4yd33Du5lo") # or get_portainer_token(base, "admin", "secret") # Example: list endpoints endpoints = api_get(base, "/api/endpoints", token) #print(endpoints) for ep in endpoints: print(f"Endpoint ID: {ep['Id']}, Name: {ep['Name']}") stacks = get_stacks(base, token) for stack in stacks: print(f"Stack ID: {stack['Id']}, Name: {stack['Name']}") stck = get_stack(base, args.stack_id, token) print(f"Found stack: ID {stck['Id']}, Name: {stck['Name']}") print(json.dumps(stck, indent=2)) uid = str(uuid.uuid4()) try: stck["AutoUpdate"]["Webhook"] = uid except: stck["AutoUpdate"] = {} try: req = { "Name": stck["Name"], "Env": stck["Env"], "AdditionalFiles": stck["AdditionalFiles"], "AutoUpdate": stck["AutoUpdate"], "repositoryURL": stck["GitConfig"]["URL"], "ReferenceName": "refs/heads/main", "composeFile": f"{stck['Name']}/docker-compose.yml", "ConfigFilePath": f"{stck['Name']}/docker-compose.yml", "repositoryAuthentication": True, "repositoryUsername": "jaydee", "repositoryPassword": "glpat-uj-n-eEfTY398PE4vKSS", "AuthorizationType": 0, "TLSSkipVerify": False, "supportRelativePath": True, "repositoryAuthentication": True, "fromAppTemplate": False, "registries": [6], "FromAppTemplate": False, "Namespace": "", "CreatedByUserId": "", "Webhook": "", "filesystemPath": "/tmp", "RegistryID": 4 } except: req = { "Name": stck["Name"], "Env": stck["Env"], "AdditionalFiles": stck["AdditionalFiles"], "AutoUpdate": None, "repositoryURL": "https://gitlab.sectorq.eu/home/docker-compose.git", "ReferenceName": "refs/heads/main", "composeFile": f"{stck['Name']}/docker-compose.yml", "ConfigFilePath": f"{stck['Name']}/docker-compose.yml", "repositoryAuthentication": True, "repositoryUsername": "jaydee", "repositoryPassword": "glpat-uj-n-eEfTY398PE4vKSS", "AuthorizationType": 0, "TLSSkipVerify": False, "supportRelativePath": True, "repositoryAuthentication": True, "fromAppTemplate": False, "registries": [6], "FromAppTemplate": False, "Namespace": "", "CreatedByUserId": "", "Webhook": "", "filesystemPath": "/tmp", "RegistryID": 6 } print(json.dumps(req, indent=2)) if args.create_stack: create_stack(base, token, 41, data=req) else: print("Not creating stack, --create-stack not provided.")