From f4c2f3b88f9496952b512b481d9620bb7856bd76 Mon Sep 17 00:00:00 2001 From: RomanSemkin Date: Tue, 10 Mar 2026 11:56:46 +0200 Subject: [PATCH 1/7] new query param for get_user's_secret as format=kubeconfig --- privx_api/api_proxy.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/privx_api/api_proxy.py b/privx_api/api_proxy.py index 7959f8e..d59bb31 100644 --- a/privx_api/api_proxy.py +++ b/privx_api/api_proxy.py @@ -258,6 +258,7 @@ def delete_current_user_client_credential( def get_current_user_client_credential_secret( self, credential_id: str, + conf_format: Optional[str] = None, ) -> PrivXAPIResponse: """ Get a current user's client credential secret. @@ -265,9 +266,11 @@ def get_current_user_client_credential_secret( Returns: PrivXAPIResponse """ + query_params = self._get_search_params(format=conf_format) response_status, data = self._http_get( UrlEnum.API_PROXY.CURRENT_CLIENT_CREDENTIAL_SECRET, path_params={"credential_id": credential_id}, + query_params=query_params, ) return self._api_response(response_status, HTTPStatus.OK, data) @@ -373,6 +376,7 @@ def get_user_client_credential_secret( self, user_id: str, credential_id: str, + conf_format: Optional[str] = None, ) -> PrivXAPIResponse: """ Fetch a user-owned client credential secret by ID. @@ -380,8 +384,10 @@ def get_user_client_credential_secret( Returns: PrivXAPIResponse """ + query_params = self._get_search_params(format=conf_format) response_status, data = self._http_get( UrlEnum.API_PROXY.USER_CLIENT_CREDENTIAL_SECRET, path_params={"user_id": user_id, "credential_id": credential_id}, + query_params=query_params, ) return self._api_response(response_status, HTTPStatus.OK, data) From 1463a44b4bc1d89ca43d9797e8b406eabb9b6976 Mon Sep 17 00:00:00 2001 From: RomanSemkin Date: Tue, 10 Mar 2026 11:57:23 +0200 Subject: [PATCH 2/7] add new query params for search connections with verbose=true/false --- privx_api/connection_manager.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/privx_api/connection_manager.py b/privx_api/connection_manager.py index f6dd02e..30023f6 100644 --- a/privx_api/connection_manager.py +++ b/privx_api/connection_manager.py @@ -53,6 +53,7 @@ def search_connections( sort_dir: Optional[str] = None, connection_params: Optional[dict] = None, fuzzy_count: Optional[bool] = False, + verbose: Optional[bool] = False, ) -> PrivXAPIResponse: """ Search for connections. @@ -66,6 +67,7 @@ def search_connections( sortkey=sort_key, sortdir=sort_dir, fuzzycount=bool(fuzzy_count), + verbose=bool(verbose), ) response_status, data = self._http_post( From db7d81e14c85954da516f65a1417347fb3de93e0 Mon Sep 17 00:00:00 2001 From: RomanSemkin Date: Tue, 10 Mar 2026 11:58:53 +0200 Subject: [PATCH 3/7] add new example of how to get a kubeconfig via privx-api-proxy --- examples/get_kubeconfig_via_api_proxy.py | 279 +++++++++++++++++++++++ 1 file changed, 279 insertions(+) create mode 100644 examples/get_kubeconfig_via_api_proxy.py diff --git a/examples/get_kubeconfig_via_api_proxy.py b/examples/get_kubeconfig_via_api_proxy.py new file mode 100644 index 0000000..784596c --- /dev/null +++ b/examples/get_kubeconfig_via_api_proxy.py @@ -0,0 +1,279 @@ +"""Generate Kubernetes kubeconfig via a PrivX API Proxy client credential. + +This example shows one practical SDK flow for Kubernetes access via PrivX: +1. List PrivX API targets or resolve a target by name. +2. Find an existing current-user client credential for that target. +3. Optionally create the credential if it does not exist yet. +4. Request the client credential secret in ``kubeconfig`` format. + +Install the SDK: +- From a local checkout: + ``pip install .`` +- Directly from GitHub: + ``pip install git+https://github.com/SSHcom/privx-sdk-for-python.git`` + +What you need before running this script: +- ``config.py`` next to this script must point to your PrivX instance and contain valid API + client credentials. +- In PrivX, the selected API target must represent the Kubernetes API you want + to access through API Proxy. +- The authenticated PrivX user must be allowed to use that target. + +Typical usage: +- List targets: + ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --list-targets`` +- Print kubeconfig to stdout: + ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --api-target my-k8s-api`` +- Create missing credential and save kubeconfig: + ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --api-target my-k8s-api --create-if-missing --output ./kubeconfig.yaml`` +- Use with kubectl: + ``KUBECONFIG=./kubeconfig.yaml kubectl get ns`` +""" + +import argparse +import os +import sys +from http import HTTPStatus +from pathlib import Path +from typing import Dict, List, Optional + +import config + +try: + import privx_api +except ImportError: + # Allow running the script from repository checkout without package install. + sys.path.append(str(Path(__file__).resolve().parent.parent)) + import privx_api + + +def build_api() -> privx_api.PrivXAPI: + """Initialize and authenticate the SDK client.""" + api = privx_api.PrivXAPI( + config.HOSTNAME, + config.HOSTPORT, + config.CA_CERT, + config.OAUTH_CLIENT_ID, + config.OAUTH_CLIENT_SECRET, + ) + api.authenticate(config.API_CLIENT_ID, config.API_CLIENT_SECRET) + return api + + +def find_api_target_by_name(api: privx_api.PrivXAPI, target_name: str) -> Dict: + """Find exactly one API target by human-readable name.""" + response = api.get_api_targets(limit=200, filter_param="") + if not response.ok: + raise RuntimeError(f"Failed to list API targets: {response.data}") + + items = response.data.get("items", []) + matches = [item for item in items if item.get("name") == target_name] + if not matches: + raise RuntimeError( + f"API target '{target_name}' not found among accessible targets." + ) + if len(matches) > 1: + ids = [item.get("id") for item in matches] + raise RuntimeError( + f"Multiple API targets named '{target_name}' found. IDs: {ids}" + ) + return matches[0] + + +def list_accessible_targets(api: privx_api.PrivXAPI) -> List[Dict]: + """Return accessible API targets for quick operator discovery.""" + response = api.get_api_targets(limit=200, filter_param="") + if not response.ok: + raise RuntimeError(f"Failed to list API targets: {response.data}") + return response.data.get("items", []) + + +def find_current_user_credential( + api: privx_api.PrivXAPI, + target_id: str, + credential_name: Optional[str] = None, +) -> Optional[Dict]: + """Find an existing current-user client credential for this target.""" + response = api.get_current_user_client_credentials(limit=200) + if not response.ok: + raise RuntimeError(f"Failed to list client credentials: {response.data}") + + items: List[Dict] = response.data.get("items", []) + for item in items: + item_target = (item.get("target") or {}).get("id") + if item_target != target_id: + continue + if credential_name and item.get("name") != credential_name: + continue + return item + return None + + +def create_current_user_credential( + api: privx_api.PrivXAPI, + target_id: str, + credential_name: str, +) -> Dict: + """Create a current-user client credential bound to the selected target.""" + payload = { + "name": credential_name, + "comment": "Generated by dev/get_kubeconfig_via_api_proxy.py", + "target": {"id": target_id}, + } + response = api.create_current_user_client_credential(payload) + if not response.ok: + raise RuntimeError(f"Failed to create client credential: {response.data}") + + credential_id = response.data.get("id") + if not credential_id: + raise RuntimeError( + "Credential was created but response did not include credential ID." + ) + return {"id": credential_id, "name": credential_name, "target": {"id": target_id}} + + +def get_kubeconfig_secret(api: privx_api.PrivXAPI, credential_id: str) -> Dict: + """Fetch secret material in kubeconfig format.""" + response = api.get_current_user_client_credential_secret( + credential_id, conf_format="kubeconfig" + ) + data = response.data + if response.status != HTTPStatus.OK: + raise RuntimeError(f"Failed to fetch kubeconfig secret: {data}") + return data + + +def save_or_print_kubeconfig(kubeconfig_yaml: str, output_path: Optional[str]) -> None: + """Write kubeconfig to a file or print to stdout.""" + if output_path: + abs_path = os.path.abspath(output_path) + os.makedirs(os.path.dirname(abs_path) or ".", exist_ok=True) + with open(abs_path, "w", encoding="utf-8") as handle: + handle.write(kubeconfig_yaml) + try: + os.chmod(abs_path, 0o600) + except OSError: + pass + print(f"kubeconfig written to: {abs_path}") + return + + print(kubeconfig_yaml) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Fetch Kubernetes kubeconfig through PrivX API Proxy using a current " + "user client credential." + ), + epilog=( + "SDK installation:\n" + " pip install .\n" + " pip install git+https://github.com/SSHcom/privx-sdk-for-python.git\n\n" + "Examples:\n" + " python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --list-targets\n" + " python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --api-target my-k8s-api\n" + " python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --api-target my-k8s-api --create-if-missing " + "--output ./kubeconfig.yaml\n" + " KUBECONFIG=./kubeconfig.yaml kubectl get namespaces" + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--api-target", + help="PrivX API target name that represents the Kubernetes API endpoint.", + ) + parser.add_argument( + "--list-targets", + action="store_true", + help="List accessible API targets and exit.", + ) + parser.add_argument( + "--credential-name", + default="k8s-api-proxy-credential", + help="Credential name to search/create under the current PrivX user.", + ) + parser.add_argument( + "--create-if-missing", + action="store_true", + help="Create the client credential if one does not exist.", + ) + parser.add_argument( + "--output", + help="Write kubeconfig YAML to this file path. If omitted, prints to stdout.", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + api = build_api() + + if args.list_targets: + targets = list_accessible_targets(api) + if not targets: + print("No accessible API targets found.") + return 0 + print("Accessible API targets:") + for target in targets: + print(f"- {target.get('name')} ({target.get('id')})") + return 0 + + if not args.api_target: + raise RuntimeError("--api-target is required unless --list-targets is used.") + + target = find_api_target_by_name(api, args.api_target) + print(f"Using API target: {target.get('name')} ({target.get('id')})") + + credential = find_current_user_credential( + api, + target_id=target.get("id", ""), + credential_name=args.credential_name, + ) + if credential is None: + if not args.create_if_missing: + print( + "No matching client credential found. Re-run with " + "--create-if-missing to create one." + ) + return 2 + credential = create_current_user_credential( + api, + target_id=target.get("id", ""), + credential_name=args.credential_name, + ) + print( + f"Created client credential: {credential.get('name')} " + f"({credential.get('id')})" + ) + else: + print( + f"Using existing client credential: {credential.get('name')} " + f"({credential.get('id')})" + ) + + kube_data = get_kubeconfig_secret(api, credential.get("id", "")) + kubeconfig = ((kube_data.get("kubeconfig") or {}).get("data") or "").strip() + if not kubeconfig: + raise RuntimeError( + "PrivX response did not contain kubeconfig.data. " + f"Full response: {kube_data}" + ) + + save_or_print_kubeconfig(kubeconfig, args.output) + + kubectl_commands = (kube_data.get("kubectl_commands") or {}).get("data") or [] + if kubectl_commands: + print("\nSuggested kubectl commands:") + for command in kubectl_commands: + print(f"- {command}") + + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: # pragma: no cover - dev script error path + print(f"ERROR: {exc}", file=sys.stderr) + raise SystemExit(1) From 41c09be931bba4fb6af0f2a1827bd27b8601c1f8 Mon Sep 17 00:00:00 2001 From: RomanSemkin Date: Tue, 10 Mar 2026 11:59:50 +0200 Subject: [PATCH 4/7] bump sdk version to 43 --- setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/setup.py b/setup.py index e0b22b1..8293165 100644 --- a/setup.py +++ b/setup.py @@ -4,7 +4,7 @@ setup( name="privx_api", - version="42.0.1", + version="43.0.0", packages=["privx_api"], license="Apache Licence 2.0", url="https://github.com/SSHcom/privx-sdk-for-python", From 1e238334b0fe2bb79acea326f3eb433d53c76843 Mon Sep 17 00:00:00 2001 From: RomanSemkin Date: Tue, 10 Mar 2026 12:04:47 +0200 Subject: [PATCH 5/7] add actions for pull requests into release branches --- .github/workflows/pythonpackage.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/pythonpackage.yml b/.github/workflows/pythonpackage.yml index 68d2a59..6956c4d 100644 --- a/.github/workflows/pythonpackage.yml +++ b/.github/workflows/pythonpackage.yml @@ -7,7 +7,7 @@ on: push: branches: [ master, main, 'v*' ] pull_request: - branches: [ master, main ] + branches: [ master, main, 'v*' ] jobs: build: From 2bf04e561f68c4e940a951a1b5787469b35194b8 Mon Sep 17 00:00:00 2001 From: RomanSemkin Date: Tue, 10 Mar 2026 12:10:14 +0200 Subject: [PATCH 6/7] fix linter's complaints --- examples/get_kubeconfig_via_api_proxy.py | 31 +++++++++++++++--------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/examples/get_kubeconfig_via_api_proxy.py b/examples/get_kubeconfig_via_api_proxy.py index 784596c..ea415cb 100644 --- a/examples/get_kubeconfig_via_api_proxy.py +++ b/examples/get_kubeconfig_via_api_proxy.py @@ -1,4 +1,5 @@ -"""Generate Kubernetes kubeconfig via a PrivX API Proxy client credential. +"""Generate Kubernetes kubeconfig via a PrivX API Proxy client +credential. This example shows one practical SDK flow for Kubernetes access via PrivX: 1. List PrivX API targets or resolve a target by name. @@ -13,19 +14,22 @@ ``pip install git+https://github.com/SSHcom/privx-sdk-for-python.git`` What you need before running this script: -- ``config.py`` next to this script must point to your PrivX instance and contain valid API - client credentials. +- ``config.py`` next to this script must point to your PrivX instance and + contain valid API client credentials. - In PrivX, the selected API target must represent the Kubernetes API you want to access through API Proxy. - The authenticated PrivX user must be allowed to use that target. Typical usage: - List targets: - ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --list-targets`` + ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py`` + ``--list-targets`` - Print kubeconfig to stdout: - ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --api-target my-k8s-api`` + ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py`` + ``--api-target my-k8s-api`` - Create missing credential and save kubeconfig: - ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --api-target my-k8s-api --create-if-missing --output ./kubeconfig.yaml`` + ``python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py`` + ``--api-target my-k8s-api --create-if-missing --output ./kubeconfig.yaml`` - Use with kubectl: ``KUBECONFIG=./kubeconfig.yaml kubectl get ns`` """ @@ -117,7 +121,7 @@ def create_current_user_credential( """Create a current-user client credential bound to the selected target.""" payload = { "name": credential_name, - "comment": "Generated by dev/get_kubeconfig_via_api_proxy.py", + "comment": "Generated by get_kubeconfig_via_api_proxy.py", "target": {"id": target_id}, } response = api.create_current_user_client_credential(payload) @@ -169,11 +173,16 @@ def parse_args() -> argparse.Namespace: epilog=( "SDK installation:\n" " pip install .\n" - " pip install git+https://github.com/SSHcom/privx-sdk-for-python.git\n\n" + " pip install " + "git+https://github.com/SSHcom/privx-sdk-for-python.git\n\n" "Examples:\n" - " python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --list-targets\n" - " python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --api-target my-k8s-api\n" - " python3 your_path_to_the_script/get_kubeconfig_via_api_proxy.py --api-target my-k8s-api --create-if-missing " + " python3 your_path_to_the_script/" + "get_kubeconfig_via_api_proxy.py --list-targets\n" + " python3 your_path_to_the_script/" + "get_kubeconfig_via_api_proxy.py --api-target my-k8s-api\n" + " python3 your_path_to_the_script/" + "get_kubeconfig_via_api_proxy.py --api-target my-k8s-api " + "--create-if-missing " "--output ./kubeconfig.yaml\n" " KUBECONFIG=./kubeconfig.yaml kubectl get namespaces" ), From c9aefe1e57b004def6f506a751a02bf374c969c9 Mon Sep 17 00:00:00 2001 From: RomanSemkin Date: Tue, 17 Mar 2026 11:04:06 +0200 Subject: [PATCH 7/7] bump linters version to black 26.3.1 --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 1e5fa12..775baf7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -black==24.3.0 +black==26.3.1 flake8-polyfill==1.0.2 flake8==5.0.4 isort==5.8.0