CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-kubernetes

Python client library for interacting with Kubernetes clusters through the Kubernetes API

—

Pending
Overview
Eval results
Files

rbac-security.mddocs/

RBAC and Security

Role-based access control management including roles, cluster roles, role bindings, service accounts, and security policies. Essential for implementing proper access controls and security policies in Kubernetes clusters with fine-grained permission management.

Capabilities

Role Management

Create and manage roles that define permissions within specific namespaces.

class RbacAuthorizationV1Api:
    def create_namespaced_role(
        self,
        namespace: str,
        body: V1Role,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1Role:
        """Create a role in specified namespace."""
    
    def list_namespaced_role(
        self,
        namespace: str,
        pretty: str = None,
        allow_watch_bookmarks: bool = None,
        continue_: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        resource_version: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1RoleList:
        """List roles in specified namespace."""
    
    def read_namespaced_role(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V1Role:
        """Read specified role."""
    
    def patch_namespaced_role(
        self,
        name: str,
        namespace: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        force: bool = None,
        pretty: str = None
    ) -> V1Role:
        """Patch specified role."""
    
    def delete_namespaced_role(
        self,
        name: str,
        namespace: str,
        body: V1DeleteOptions = None,
        dry_run: str = None,
        grace_period_seconds: int = None,
        orphan_dependents: bool = None,
        propagation_policy: str = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified role."""

Cluster Role Management

Create and manage cluster-wide roles that define permissions across the entire cluster.

class RbacAuthorizationV1Api:
    def create_cluster_role(
        self,
        body: V1ClusterRole,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1ClusterRole:
        """Create a cluster role."""
    
    def list_cluster_role(
        self,
        pretty: str = None,
        allow_watch_bookmarks: bool = None,
        continue_: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        resource_version: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1ClusterRoleList:
        """List cluster roles."""
    
    def read_cluster_role(
        self,
        name: str,
        pretty: str = None
    ) -> V1ClusterRole:
        """Read specified cluster role."""

Role Binding Management

Bind roles to users, groups, or service accounts to grant permissions.

class RbacAuthorizationV1Api:
    def create_namespaced_role_binding(
        self,
        namespace: str,
        body: V1RoleBinding,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1RoleBinding:
        """Create a role binding in specified namespace."""
    
    def list_namespaced_role_binding(
        self,
        namespace: str,
        pretty: str = None,
        allow_watch_bookmarks: bool = None,
        continue_: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        resource_version: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1RoleBindingList:
        """List role bindings in specified namespace."""
    
    def create_cluster_role_binding(
        self,
        body: V1ClusterRoleBinding,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1ClusterRoleBinding:
        """Create a cluster role binding."""
    
    def list_cluster_role_binding(
        self,
        pretty: str = None,
        allow_watch_bookmarks: bool = None,
        continue_: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        resource_version: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1ClusterRoleBindingList:
        """List cluster role bindings."""

Service Account Management

Create and manage service accounts for pod authentication and authorization.

class CoreV1Api:
    def create_namespaced_service_account(
        self,
        namespace: str,
        body: V1ServiceAccount,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1ServiceAccount:
        """Create a service account in specified namespace."""
    
    def list_namespaced_service_account(
        self,
        namespace: str,
        pretty: str = None,
        allow_watch_bookmarks: bool = None,
        continue_: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        resource_version: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1ServiceAccountList:
        """List service accounts in specified namespace."""
    
    def read_namespaced_service_account(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V1ServiceAccount:
        """Read specified service account."""

Resource Models

V1Role

class V1Role:
    api_version: str  # "rbac.authorization.k8s.io/v1"
    kind: str  # "Role"
    metadata: V1ObjectMeta
    rules: list  # List of V1PolicyRule

V1ClusterRole

class V1ClusterRole:
    api_version: str  # "rbac.authorization.k8s.io/v1"
    kind: str  # "ClusterRole"
    metadata: V1ObjectMeta
    rules: list  # List of V1PolicyRule
    aggregation_rule: V1AggregationRule

V1RoleBinding

class V1RoleBinding:
    api_version: str  # "rbac.authorization.k8s.io/v1"
    kind: str  # "RoleBinding"
    metadata: V1ObjectMeta
    subjects: list  # List of V1Subject
    role_ref: V1RoleRef

V1ClusterRoleBinding

class V1ClusterRoleBinding:
    api_version: str  # "rbac.authorization.k8s.io/v1"
    kind: str  # "ClusterRoleBinding"
    metadata: V1ObjectMeta
    subjects: list  # List of V1Subject
    role_ref: V1RoleRef

V1ServiceAccount

class V1ServiceAccount:
    api_version: str  # "v1"
    kind: str  # "ServiceAccount"
    metadata: V1ObjectMeta
    secrets: list  # List of V1ObjectReference
    image_pull_secrets: list  # List of V1LocalObjectReference
    automount_service_account_token: bool

V1PolicyRule

class V1PolicyRule:
    verbs: list  # ["get", "list", "create", "update", "patch", "delete"]
    api_groups: list  # ["", "apps", "extensions"]
    resources: list  # ["pods", "deployments", "services"]
    resource_names: list  # Specific resource names
    non_resource_urls: list  # Non-resource URLs like /healthz

V1Subject

class V1Subject:
    kind: str  # "User", "Group", "ServiceAccount"
    name: str
    namespace: str  # For ServiceAccount subjects
    api_group: str  # "rbac.authorization.k8s.io" for User/Group

V1RoleRef

class V1RoleRef:
    api_group: str  # "rbac.authorization.k8s.io"
    kind: str  # "Role" or "ClusterRole"
    name: str

Usage Examples

Creating a Role

from kubernetes import client, config

config.load_kube_config()
rbac_api = client.RbacAuthorizationV1Api()

# Define role with specific permissions
role_manifest = {
    "apiVersion": "rbac.authorization.k8s.io/v1",
    "kind": "Role",
    "metadata": {
        "name": "pod-reader",
        "namespace": "default"
    },
    "rules": [{
        "apiGroups": [""],
        "resources": ["pods"],
        "verbs": ["get", "list"]
    }, {
        "apiGroups": [""],
        "resources": ["pods/log"],
        "verbs": ["get"]
    }]
}

# Create the role
try:
    role = rbac_api.create_namespaced_role(
        namespace="default",
        body=role_manifest
    )
    print(f"Role created: {role.metadata.name}")
except client.ApiException as e:
    print(f"Failed to create role: {e}")

Creating a Cluster Role

from kubernetes import client, config

config.load_kube_config()
rbac_api = client.RbacAuthorizationV1Api()

# Define cluster role with node management permissions
cluster_role_manifest = {
    "apiVersion": "rbac.authorization.k8s.io/v1",
    "kind": "ClusterRole",
    "metadata": {
        "name": "node-manager"
    },
    "rules": [{
        "apiGroups": [""],
        "resources": ["nodes"],
        "verbs": ["get", "list", "patch", "update"]
    }, {
        "apiGroups": [""],
        "resources": ["nodes/status"],
        "verbs": ["get", "update"]
    }, {
        "apiGroups": ["metrics.k8s.io"],
        "resources": ["nodes", "pods"],
        "verbs": ["get", "list"]
    }]
}

# Create cluster role
cluster_role = rbac_api.create_cluster_role(body=cluster_role_manifest)
print(f"Cluster role created: {cluster_role.metadata.name}")

Creating Service Account and Role Binding

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()
rbac_api = client.RbacAuthorizationV1Api()

# Create service account
service_account = {
    "apiVersion": "v1",
    "kind": "ServiceAccount",
    "metadata": {
        "name": "app-service-account",
        "namespace": "default"
    }
}

sa = v1.create_namespaced_service_account(
    namespace="default",
    body=service_account
)
print(f"Service account created: {sa.metadata.name}")

# Create role binding to bind role to service account
role_binding = {
    "apiVersion": "rbac.authorization.k8s.io/v1",
    "kind": "RoleBinding",
    "metadata": {
        "name": "app-pod-reader-binding",
        "namespace": "default"
    },
    "subjects": [{
        "kind": "ServiceAccount",
        "name": "app-service-account",
        "namespace": "default"
    }],
    "roleRef": {
        "kind": "Role",
        "name": "pod-reader",
        "apiGroup": "rbac.authorization.k8s.io"
    }
}

rb = rbac_api.create_namespaced_role_binding(
    namespace="default",
    body=role_binding
)
print(f"Role binding created: {rb.metadata.name}")

Creating Cluster Role Binding

from kubernetes import client, config

config.load_kube_config()
rbac_api = client.RbacAuthorizationV1Api()

# Bind cluster role to user
cluster_role_binding = {
    "apiVersion": "rbac.authorization.k8s.io/v1",
    "kind": "ClusterRoleBinding",
    "metadata": {
        "name": "node-manager-binding"
    },
    "subjects": [{
        "kind": "User",
        "name": "admin@mycompany.com",
        "apiGroup": "rbac.authorization.k8s.io"
    }, {
        "kind": "ServiceAccount",
        "name": "node-operator",
        "namespace": "kube-system"
    }],
    "roleRef": {
        "kind": "ClusterRole",
        "name": "node-manager",
        "apiGroup": "rbac.authorization.k8s.io"
    }
}

crb = rbac_api.create_cluster_role_binding(body=cluster_role_binding)
print(f"Cluster role binding created: {crb.metadata.name}")

Creating Comprehensive RBAC Setup

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()
rbac_api = client.RbacAuthorizationV1Api()

def create_rbac_setup(namespace, app_name, permissions):
    """Create complete RBAC setup for an application."""
    
    # 1. Create service account
    service_account = {
        "apiVersion": "v1",
        "kind": "ServiceAccount",
        "metadata": {
            "name": f"{app_name}-sa",
            "namespace": namespace,
            "labels": {"app": app_name}
        }
    }
    
    sa = v1.create_namespaced_service_account(
        namespace=namespace,
        body=service_account
    )
    print(f"✓ Created service account: {sa.metadata.name}")
    
    # 2. Create role with specified permissions
    role = {
        "apiVersion": "rbac.authorization.k8s.io/v1",
        "kind": "Role",
        "metadata": {
            "name": f"{app_name}-role",
            "namespace": namespace,
            "labels": {"app": app_name}
        },
        "rules": permissions
    }
    
    r = rbac_api.create_namespaced_role(
        namespace=namespace,
        body=role
    )
    print(f"✓ Created role: {r.metadata.name}")
    
    # 3. Create role binding
    role_binding = {
        "apiVersion": "rbac.authorization.k8s.io/v1",
        "kind": "RoleBinding",
        "metadata": {
            "name": f"{app_name}-binding",
            "namespace": namespace,
            "labels": {"app": app_name}
        },
        "subjects": [{
            "kind": "ServiceAccount",
            "name": f"{app_name}-sa",
            "namespace": namespace
        }],
        "roleRef": {
            "kind": "Role",
            "name": f"{app_name}-role",
            "apiGroup": "rbac.authorization.k8s.io"
        }
    }
    
    rb = rbac_api.create_namespaced_role_binding(
        namespace=namespace,
        body=role_binding
    )
    print(f"✓ Created role binding: {rb.metadata.name}")
    
    return sa, r, rb

# Define permissions for a monitoring application
monitoring_permissions = [{
    "apiGroups": [""],
    "resources": ["pods", "services", "endpoints"],
    "verbs": ["get", "list", "watch"]
}, {
    "apiGroups": [""],
    "resources": ["configmaps"],
    "verbs": ["get", "list"]
}, {
    "apiGroups": ["apps"],
    "resources": ["deployments", "replicasets"],
    "verbs": ["get", "list", "watch"]
}]

# Create RBAC setup
sa, role, binding = create_rbac_setup(
    "monitoring",
    "prometheus",
    monitoring_permissions
)

Working with Built-in Cluster Roles

from kubernetes import client, config

config.load_kube_config()
rbac_api = client.RbacAuthorizationV1Api()

# List built-in cluster roles
cluster_roles = rbac_api.list_cluster_role()

print("Built-in cluster roles:")
builtin_roles = []
for role in cluster_roles.items:
    if role.metadata.name.startswith('system:') or role.metadata.name in [
        'admin', 'edit', 'view', 'cluster-admin'
    ]:
        builtin_roles.append(role.metadata.name)

for role_name in sorted(builtin_roles):
    print(f"  - {role_name}")

# Use built-in 'view' role for read-only access
readonly_binding = {
    "apiVersion": "rbac.authorization.k8s.io/v1",
    "kind": "RoleBinding",
    "metadata": {
        "name": "developers-readonly",
        "namespace": "development"
    },
    "subjects": [{
        "kind": "Group",
        "name": "developers",
        "apiGroup": "rbac.authorization.k8s.io"
    }],
    "roleRef": {
        "kind": "ClusterRole",
        "name": "view",
        "apiGroup": "rbac.authorization.k8s.io"
    }
}

rb = rbac_api.create_namespaced_role_binding(
    namespace="development",
    body=readonly_binding
)
print(f"✓ Created read-only binding: {rb.metadata.name}")

Authorization Checking

from kubernetes import client, config

config.load_kube_config()
auth_api = client.AuthorizationV1Api()

def can_i(verb, resource, namespace=None, group="", name=None):
    """Check if current user can perform action on resource."""
    
    resource_attributes = {
        "verb": verb,
        "group": group,
        "resource": resource
    }
    
    if namespace:
        resource_attributes["namespace"] = namespace
    if name:
        resource_attributes["name"] = name
    
    # Create SelfSubjectAccessReview
    access_review = {
        "apiVersion": "authorization.k8s.io/v1",
        "kind": "SelfSubjectAccessReview",
        "spec": {
            "resourceAttributes": resource_attributes
        }
    }
    
    result = auth_api.create_self_subject_access_review(body=access_review)
    return result.status.allowed

# Check permissions
permissions_to_check = [
    ("get", "pods", "default"),
    ("create", "deployments", "default", "apps"),
    ("delete", "services", "production"),
    ("get", "nodes", None)  # Cluster-scoped
]

print("Permission check results:")
for check in permissions_to_check:
    verb, resource = check[:2]
    namespace = check[2] if len(check) > 2 else None
    group = check[3] if len(check) > 3 else ""
    
    allowed = can_i(verb, resource, namespace, group)
    status = "✓ ALLOWED" if allowed else "✗ DENIED"
    location = f"in {namespace}" if namespace else "cluster-wide"
    print(f"  {verb} {resource} {location}: {status}")

Pod Security Context

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

# Create pod with security context
secure_pod = {
    "apiVersion": "v1",
    "kind": "Pod",
    "metadata": {
        "name": "secure-pod",
        "namespace": "default"
    },
    "spec": {
        "serviceAccountName": "app-service-account",
        "securityContext": {
            "runAsNonRoot": True,
            "runAsUser": 1000,
            "runAsGroup": 1000,
            "fsGroup": 1000,
            "seccompProfile": {
                "type": "RuntimeDefault"
            }
        },
        "containers": [{
            "name": "app",
            "image": "nginx:1.20",
            "securityContext": {
                "allowPrivilegeEscalation": False,
                "readOnlyRootFilesystem": True,
                "capabilities": {
                    "drop": ["ALL"],
                    "add": ["NET_BIND_SERVICE"]
                }
            },
            "volumeMounts": [{
                "name": "tmp-volume",
                "mountPath": "/tmp"
            }, {
                "name": "cache-volume", 
                "mountPath": "/var/cache/nginx"
            }]
        }],
        "volumes": [{
            "name": "tmp-volume",
            "emptyDir": {}
        }, {
            "name": "cache-volume",
            "emptyDir": {}
        }]
    }
}

# Create secure pod
pod = v1.create_namespaced_pod(namespace="default", body=secure_pod)
print(f"✓ Created secure pod: {pod.metadata.name}")
print(f"  Running as user: {pod.spec.security_context.run_as_user}")
print(f"  Non-root: {pod.spec.security_context.run_as_non_root}")

Install with Tessl CLI

npx tessl i tessl/pypi-kubernetes

docs

application-workloads.md

autoscaling.md

configuration.md

core-resources.md

custom-resources.md

dynamic-client.md

index.md

leader-election.md

networking.md

rbac-security.md

resource-watching.md

storage.md

streaming-operations.md

utilities.md

tile.json