Merge pull request #129 from SSHcom/v43.0.0ff

V43.0.0

Author: rsemkin

.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:

examples/get_kubeconfig_via_api_proxy.py

@@ -0,0 +1,288 @@
+"""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 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)

privx_api/api_proxy.py

@@ -258,16 +258,19 @@ 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.
 
         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,15 +376,18 @@ 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.
 
         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)

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(

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

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",