Access Control

The access control module provides comprehensive security and permission management for workflows.

Access control package with composition-based architecture.

This package provides clean, testable access control components: - Rule evaluators for RBAC, ABAC, and hybrid strategies - Composable access control managers - Backward compatibility with existing code

class kailash.access_control.NodePermission(value)

Bases: Enum

Node-level permissions

EXECUTE = 'execute'
READ_OUTPUT = 'read_output'
WRITE_INPUT = 'write_input'
SKIP = 'skip'
MASK_OUTPUT = 'mask_output'
classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name)

Return the member matching name.

classmethod __iter__()

Return members in definition order.

classmethod __len__()

Return the number of members (no aliases)

class kailash.access_control.WorkflowPermission(value)

Bases: Enum

Workflow-level permissions

VIEW = 'view'
EXECUTE = 'execute'
MODIFY = 'modify'
DELETE = 'delete'
SHARE = 'share'
ADMIN = 'admin'
classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name)

Return the member matching name.

classmethod __iter__()

Return members in definition order.

classmethod __len__()

Return the number of members (no aliases)

class kailash.access_control.PermissionEffect(value)

Bases: Enum

Effect of a permission rule

ALLOW = 'allow'
DENY = 'deny'
CONDITIONAL = 'conditional'
classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name)

Return the member matching name.

classmethod __iter__()

Return members in definition order.

classmethod __len__()

Return the number of members (no aliases)

class kailash.access_control.PermissionRule(id: str, resource_type: str, resource_id: str, permission: WorkflowPermission | NodePermission, effect: PermissionEffect, user_id: str | None = None, role: str | None = None, tenant_id: str | None = None, conditions: dict[str, ~typing.Any]=<factory>, created_at: datetime = <factory>, created_by: str | None = None, expires_at: datetime | None = None, priority: int = 0)

Bases: object

A single permission rule defining access control policies.

Represents a single access control rule that grants or denies permissions to users for specific resources (workflows or nodes) based on their identity, roles, and contextual conditions.

Design Purpose:

Provides a flexible, declarative way to define access control policies. Supports role-based access control (RBAC), attribute-based access control (ABAC), and conditional permissions based on runtime context.

Upstream Dependencies:

  • Administrative interfaces for rule creation

  • Policy management systems

  • Configuration files or databases

Downstream Consumers:

  • AccessControlManager for rule evaluation

  • Audit systems for logging policy decisions

  • Policy analysis tools for rule validation

Usage Patterns:

  • Created by administrators to define access policies

  • Evaluated during workflow and node execution

  • Cached for performance optimization

  • Updated when policies change

Implementation Details:

Uses dataclass for efficient serialization and comparison. Supports priority-based rule ordering for conflict resolution. Includes expiration for time-limited permissions. Conditions enable complex policy logic.

Example

>>> rule = PermissionRule(
...     id="allow_analysts_read",
...     resource_type="node",
...     resource_id="sensitive_data",
...     permission=NodePermission.READ_OUTPUT,
...     effect=PermissionEffect.ALLOW,
...     role="analyst"
... )
>>> print(rule.id)
allow_analysts_read
Parameters:
__init__(id: str, resource_type: str, resource_id: str, permission: WorkflowPermission | NodePermission, effect: PermissionEffect, user_id: str | None = None, role: str | None = None, tenant_id: str | None = None, conditions: dict[str, ~typing.Any]=<factory>, created_at: datetime = <factory>, created_by: str | None = None, expires_at: datetime | None = None, priority: int = 0) → None
Parameters:
Return type:

None

created_by: str | None = None
expires_at: datetime | None = None
priority: int = 0
role: str | None = None
tenant_id: str | None = None
user_id: str | None = None
id: str
resource_type: str
resource_id: str
permission: WorkflowPermission | NodePermission
effect: PermissionEffect
conditions: dict[str, Any]
created_at: datetime
class kailash.access_control.UserContext(user_id: str, tenant_id: str, email: str, roles: list[str] = <factory>, permissions: list[str] = <factory>, attributes: dict[str, ~typing.Any]=<factory>, session_id: str | None = None, ip_address: str | None = None)

Bases: object

User context for access control decisions.

Contains all information needed to make access control decisions for a user, including identity, tenant membership, roles, and session information.

Design Purpose:

Provides a standardized way to represent user identity and permissions across the entire access control system. Enables fine-grained access control based on user attributes, roles, and context.

Upstream Dependencies:

  • Authentication systems (JWT, API keys)

  • User management systems

  • Tenant management systems

Downstream Consumers:

  • AccessControlManager for permission checking

  • AccessControlledRuntime for workflow execution

  • Audit logging systems for tracking access

Usage Patterns:

  • Created during authentication/login process

  • Passed to all access control functions

  • Used in workflow and node execution contexts

  • Logged for audit and compliance purposes

Implementation Details:

Uses dataclass for efficient attribute access and comparison. Immutable once created to prevent privilege escalation. Supports custom attributes for extensible authorization.

Example

>>> user = UserContext(
...     user_id="user123",
...     tenant_id="tenant001",
...     email="user@example.com",
...     roles=["analyst", "viewer"]
... )
>>> print(user.user_id)
user123
Parameters:
__init__(user_id: str, tenant_id: str, email: str, roles: list[str] = <factory>, permissions: list[str] = <factory>, attributes: dict[str, ~typing.Any]=<factory>, session_id: str | None = None, ip_address: str | None = None) → None
Parameters:
Return type:

None

ip_address: str | None = None
session_id: str | None = None
user_id: str
tenant_id: str
email: str
roles: list[str]
permissions: list[str]
attributes: dict[str, Any]
class kailash.access_control.AccessDecision(allowed: bool, reason: str, applied_rules: list[PermissionRule] = <factory>, conditions_met: dict[str, bool]=<factory>, masked_fields: list[str] = <factory>, redirect_node: str | None = None)

Bases: object

Result of an access control decision.

Contains the outcome of evaluating access control rules for a specific user and resource, including whether access is allowed, the reasoning, and any additional actions required (like data masking).

Design Purpose:

Provides a comprehensive result object that captures not just the allow/deny decision, but also the context and reasoning behind it. Enables audit logging, debugging, and conditional execution.

Upstream Dependencies:

  • AccessControlManager rule evaluation

  • Permission rule matching logic

  • Conditional evaluation systems

Downstream Consumers:

  • AccessControlledRuntime for execution decisions

  • Audit logging systems for compliance

  • Error handling for access denial messages

  • Data masking systems for output filtering

Usage Patterns:

  • Returned by all access control check methods

  • Logged for audit and debugging purposes

  • Used to determine execution flow

  • Provides user-friendly error messages

Implementation Details:

Immutable after creation to ensure decision integrity. Includes applied rules for transparency and debugging. Supports conditional decisions for complex scenarios. Contains masking information for data protection.

Example

>>> decision = AccessDecision(
...     allowed=True,
...     reason="User has analyst role",
...     masked_fields=["ssn", "phone"]
... )
>>> print(decision.allowed)
True
Parameters:
__init__(allowed: bool, reason: str, applied_rules: list[PermissionRule] = <factory>, conditions_met: dict[str, bool]=<factory>, masked_fields: list[str] = <factory>, redirect_node: str | None = None) → None
Parameters:
Return type:

None

redirect_node: str | None = None
allowed: bool
reason: str
applied_rules: list[PermissionRule]
conditions_met: dict[str, bool]
masked_fields: list[str]
class kailash.access_control.ConditionEvaluator

Bases: object

Evaluates conditions for conditional permissions

__init__()
evaluate(condition_type: str, condition_value: Any, context: dict[str, Any]) → bool

Evaluate a condition

Parameters:
Return type:

bool

class kailash.access_control.AccessControlManager(rule_evaluator: RuleEvaluator | None = None, strategy: str = 'hybrid', enabled: bool = True)[source]

Bases: object

Access control manager using composition pattern.

This manager separates rule storage from rule evaluation, allowing: - Easy testing with mock evaluators - Flexible evaluation strategies (RBAC, ABAC, Hybrid) - Clear separation of concerns - No inheritance-related bugs

Example

>>> # Create with hybrid evaluation (RBAC + ABAC)
>>> manager = AccessControlManager()

>>> # Or specify evaluation strategy
>>> rbac_manager = AccessControlManager(strategy="rbac")
>>> abac_manager = AccessControlManager(strategy="abac")

>>> # Add rules
>>> manager.add_rule(PermissionRule(...))

>>> # Check access
>>> decision = manager.check_node_access(user, "node_id", NodePermission.EXECUTE)
Parameters:

  • rule_evaluator (RuleEvaluator | None)

  • strategy (str)

  • enabled (bool)

__init__(rule_evaluator: RuleEvaluator | None = None, strategy: str = 'hybrid', enabled: bool = True)[source]

Initialize access control manager.

Parameters:

  • rule_evaluator (RuleEvaluator | None) – Custom rule evaluator (overrides strategy)

  • strategy (str) – Evaluation strategy (‘rbac’, ‘abac’, ‘hybrid’)

  • enabled (bool) – Whether access control is enabled

add_rule(rule: PermissionRule) → None[source]

Add a permission rule.

Parameters:

rule (PermissionRule) – Permission rule to add

Return type:

None

remove_rule(rule_id: str) → bool[source]

Remove a permission rule.

Parameters:

rule_id (str) – ID of rule to remove

Returns:

True if rule was found and removed

Return type:

bool

check_workflow_access(user: UserContext, workflow_id: str, permission: WorkflowPermission, runtime_context: Dict[str, Any] | None = None) → AccessDecision[source]

Check if user has permission on workflow.

Parameters:

  • user (UserContext) – User requesting access

  • workflow_id (str) – Workflow to access

  • permission (WorkflowPermission) – Permission being requested

  • runtime_context (Dict[str, Any] | None) – Additional runtime context

Returns:

AccessDecision with allow/deny and reasoning

Return type:

AccessDecision

check_node_access(user: UserContext, node_id: str, permission: NodePermission, runtime_context: Dict[str, Any] | None = None) → AccessDecision[source]

Check if user has permission on node.

Parameters:

  • user (UserContext) – User requesting access

  • node_id (str) – Node to access

  • permission (NodePermission) – Permission being requested

  • runtime_context (Dict[str, Any] | None) – Additional runtime context

Returns:

AccessDecision with allow/deny and reasoning

Return type:

AccessDecision

get_accessible_nodes(user: UserContext, workflow_id: str, permission: NodePermission) → set[str][source]

Get all nodes user can access in a workflow.

Parameters:

  • user (UserContext) – User to check access for

  • workflow_id (str) – Workflow containing nodes

  • permission (NodePermission) – Permission type to check

Returns:

Set of accessible node IDs

Return type:

set[str]

add_masking_rule(node_id: str, rule: Any) → None[source]

Add attribute-based masking rule for a node.

Parameters:
Return type:

None

apply_data_masking(user: UserContext, node_id: str, data: Dict[str, Any]) → Dict[str, Any][source]

Apply attribute-based data masking to node output.

Parameters:
Return type:

Dict[str, Any]

supports_conditions() → bool[source]

Check if current evaluator supports conditional rules.

Returns:

True if complex conditions are supported

Return type:

bool

get_strategy_info() → Dict[str, Any][source]

Get information about the current evaluation strategy.

Returns:

Dictionary with strategy details

Return type:

Dict[str, Any]

class kailash.access_control.RuleEvaluator[source]

Bases: ABC

Abstract base class for rule evaluation strategies.

abstractmethod evaluate(rules: List[PermissionRule], user: UserContext, resource_type: str, resource_id: str, permission: NodePermission | WorkflowPermission, runtime_context: Dict[str, Any]) → AccessDecision[source]

Evaluate a set of rules for a user’s access request.

Parameters:
Returns:

AccessDecision with allow/deny and reasoning

Return type:

AccessDecision

evaluate_rules(*args, **kwargs)[source]
abstractmethod supports_conditions() → bool[source]

Return whether this evaluator supports conditional rules.

Return type:

bool

class kailash.access_control.RBACRuleEvaluator[source]

Bases: RuleEvaluator

Role-Based Access Control rule evaluator.

This evaluator handles traditional RBAC rules based on: - User roles - Direct user assignments - Tenant-based permissions

__init__()[source]

Initialize RBAC evaluator.

evaluate(rules: List[PermissionRule], user: UserContext, resource_type: str, resource_id: str, permission: NodePermission | WorkflowPermission, runtime_context: Dict[str, Any]) → AccessDecision[source]

Evaluate RBAC rules.

Parameters:
Return type:

AccessDecision

supports_conditions() → bool[source]

RBAC evaluator does not support complex conditions.

Return type:

bool

evaluate_rules(*args, **kwargs)
class kailash.access_control.ABACRuleEvaluator[source]

Bases: RuleEvaluator

Attribute-Based Access Control rule evaluator.

This evaluator handles advanced ABAC rules with: - Complex attribute expressions - Dynamic condition evaluation - Hierarchical attribute matching

__init__()[source]

Initialize ABAC evaluator.

evaluate(rules: List[PermissionRule], user: UserContext, resource_type: str, resource_id: str, permission: NodePermission | WorkflowPermission, runtime_context: Dict[str, Any]) → AccessDecision[source]

Evaluate ABAC rules with complex attribute conditions.

Parameters:
Return type:

AccessDecision

supports_conditions() → bool[source]

ABAC evaluator fully supports complex conditions.

Return type:

bool

evaluate_rules(*args, **kwargs)
class kailash.access_control.HybridRuleEvaluator[source]

Bases: RuleEvaluator

Hybrid evaluator that combines RBAC and ABAC evaluation.

This evaluator: 1. Uses RBAC for basic rules without conditions 2. Uses ABAC for complex conditional rules 3. Provides seamless transition between evaluation strategies

__init__()[source]

Initialize hybrid evaluator with both RBAC and ABAC.

evaluate(rules: List[PermissionRule], user: UserContext, resource_type: str, resource_id: str, permission: NodePermission | WorkflowPermission, runtime_context: Dict[str, Any]) → AccessDecision[source]

Evaluate rules using appropriate strategy based on rule complexity.

Parameters:
Return type:

AccessDecision

supports_conditions() → bool[source]

Hybrid evaluator supports conditions via ABAC component.

Return type:

bool

evaluate_rules(*args, **kwargs)
kailash.access_control.create_rule_evaluator(strategy: str = 'hybrid') → RuleEvaluator[source]

Create a rule evaluator based on strategy.

Parameters:

strategy (str) – One of ‘rbac’, ‘abac’, or ‘hybrid’

Returns:

Appropriate RuleEvaluator instance

Raises:

ValueError – If strategy is not recognized

Return type:

RuleEvaluator

kailash.access_control.get_access_control_manager() → AccessControlManager

Get the global access control manager

Return type:

AccessControlManager

kailash.access_control.set_access_control_manager(manager: AccessControlManager)

Set a custom access control manager

Parameters:

manager (AccessControlManager)

kailash.access_control.create_attribute_condition(path: str, operator: str, value: Any, case_sensitive: bool = True) → Dict[str, Any][source]

Create an attribute condition configuration.

Helper function to create properly formatted attribute conditions for use in permission rules.

Parameters:

  • path (str) – Attribute path (e.g., “user.attributes.department”)

  • operator (str) – Comparison operator

  • value (Any) – Value to compare against

  • case_sensitive (bool) – Whether comparison is case sensitive

Returns:

Condition configuration dict

Return type:

Dict[str, Any]

kailash.access_control.create_complex_condition(operator: str, conditions: List[Dict[str, Any]]) → Dict[str, Any][source]

Create a complex attribute condition with logical operators.

Helper function to create AND/OR/NOT conditions.

Parameters:

  • operator (str) – Logical operator (and/or/not)

  • conditions (List[Dict[str, Any]]) – List of conditions to combine

Returns:

Complex condition configuration dict

Return type:

Dict[str, Any]

Access Control Manager

class kailash.access_control.AccessControlManager(rule_evaluator: RuleEvaluator | None = None, strategy: str = 'hybrid', enabled: bool = True)[source]

Bases: object

Access control manager using composition pattern.

This manager separates rule storage from rule evaluation, allowing: - Easy testing with mock evaluators - Flexible evaluation strategies (RBAC, ABAC, Hybrid) - Clear separation of concerns - No inheritance-related bugs

Example

>>> # Create with hybrid evaluation (RBAC + ABAC)
>>> manager = AccessControlManager()

>>> # Or specify evaluation strategy
>>> rbac_manager = AccessControlManager(strategy="rbac")
>>> abac_manager = AccessControlManager(strategy="abac")

>>> # Add rules
>>> manager.add_rule(PermissionRule(...))

>>> # Check access
>>> decision = manager.check_node_access(user, "node_id", NodePermission.EXECUTE)
Parameters:

  • rule_evaluator (RuleEvaluator | None)

  • strategy (str)

  • enabled (bool)

__init__(rule_evaluator: RuleEvaluator | None = None, strategy: str = 'hybrid', enabled: bool = True)[source]

Initialize access control manager.

Parameters:

  • rule_evaluator (RuleEvaluator | None) – Custom rule evaluator (overrides strategy)

  • strategy (str) – Evaluation strategy (‘rbac’, ‘abac’, ‘hybrid’)

  • enabled (bool) – Whether access control is enabled

add_rule(rule: PermissionRule) → None[source]

Add a permission rule.

Parameters:

rule (PermissionRule) – Permission rule to add

Return type:

None

remove_rule(rule_id: str) → bool[source]

Remove a permission rule.

Parameters:

rule_id (str) – ID of rule to remove

Returns:

True if rule was found and removed

Return type:

bool

check_workflow_access(user: UserContext, workflow_id: str, permission: WorkflowPermission, runtime_context: Dict[str, Any] | None = None) → AccessDecision[source]

Check if user has permission on workflow.

Parameters:

  • user (UserContext) – User requesting access

  • workflow_id (str) – Workflow to access

  • permission (WorkflowPermission) – Permission being requested

  • runtime_context (Dict[str, Any] | None) – Additional runtime context

Returns:

AccessDecision with allow/deny and reasoning

Return type:

AccessDecision

check_node_access(user: UserContext, node_id: str, permission: NodePermission, runtime_context: Dict[str, Any] | None = None) → AccessDecision[source]

Check if user has permission on node.

Parameters:

  • user (UserContext) – User requesting access

  • node_id (str) – Node to access

  • permission (NodePermission) – Permission being requested

  • runtime_context (Dict[str, Any] | None) – Additional runtime context

Returns:

AccessDecision with allow/deny and reasoning

Return type:

AccessDecision

get_accessible_nodes(user: UserContext, workflow_id: str, permission: NodePermission) → set[str][source]

Get all nodes user can access in a workflow.

Parameters:

  • user (UserContext) – User to check access for

  • workflow_id (str) – Workflow containing nodes

  • permission (NodePermission) – Permission type to check

Returns:

Set of accessible node IDs

Return type:

set[str]

add_masking_rule(node_id: str, rule: Any) → None[source]

Add attribute-based masking rule for a node.

Parameters:
Return type:

None

apply_data_masking(user: UserContext, node_id: str, data: Dict[str, Any]) → Dict[str, Any][source]

Apply attribute-based data masking to node output.

Parameters:
Return type:

Dict[str, Any]

supports_conditions() → bool[source]

Check if current evaluator supports conditional rules.

Returns:

True if complex conditions are supported

Return type:

bool

get_strategy_info() → Dict[str, Any][source]

Get information about the current evaluation strategy.

Returns:

Dictionary with strategy details

Return type:

Dict[str, Any]

User Context

class kailash.access_control.UserContext(user_id: str, tenant_id: str, email: str, roles: list[str] = <factory>, permissions: list[str] = <factory>, attributes: dict[str, ~typing.Any]=<factory>, session_id: str | None = None, ip_address: str | None = None)

Bases: object

User context for access control decisions.

Contains all information needed to make access control decisions for a user, including identity, tenant membership, roles, and session information.

Design Purpose:

Provides a standardized way to represent user identity and permissions across the entire access control system. Enables fine-grained access control based on user attributes, roles, and context.

Upstream Dependencies:

  • Authentication systems (JWT, API keys)

  • User management systems

  • Tenant management systems

Downstream Consumers:

  • AccessControlManager for permission checking

  • AccessControlledRuntime for workflow execution

  • Audit logging systems for tracking access

Usage Patterns:

  • Created during authentication/login process

  • Passed to all access control functions

  • Used in workflow and node execution contexts

  • Logged for audit and compliance purposes

Implementation Details:

Uses dataclass for efficient attribute access and comparison. Immutable once created to prevent privilege escalation. Supports custom attributes for extensible authorization.

Example

>>> user = UserContext(
...     user_id="user123",
...     tenant_id="tenant001",
...     email="user@example.com",
...     roles=["analyst", "viewer"]
... )
>>> print(user.user_id)
user123
Parameters:
__init__(user_id: str, tenant_id: str, email: str, roles: list[str] = <factory>, permissions: list[str] = <factory>, attributes: dict[str, ~typing.Any]=<factory>, session_id: str | None = None, ip_address: str | None = None) → None
Parameters:
Return type:

None

ip_address: str | None = None
session_id: str | None = None
user_id: str
tenant_id: str
email: str
roles: list[str]
permissions: list[str]
attributes: dict[str, Any]

Permission Rules

class kailash.access_control.PermissionRule(id: str, resource_type: str, resource_id: str, permission: WorkflowPermission | NodePermission, effect: PermissionEffect, user_id: str | None = None, role: str | None = None, tenant_id: str | None = None, conditions: dict[str, ~typing.Any]=<factory>, created_at: datetime = <factory>, created_by: str | None = None, expires_at: datetime | None = None, priority: int = 0)

Bases: object

A single permission rule defining access control policies.

Represents a single access control rule that grants or denies permissions to users for specific resources (workflows or nodes) based on their identity, roles, and contextual conditions.

Design Purpose:

Provides a flexible, declarative way to define access control policies. Supports role-based access control (RBAC), attribute-based access control (ABAC), and conditional permissions based on runtime context.

Upstream Dependencies:

  • Administrative interfaces for rule creation

  • Policy management systems

  • Configuration files or databases

Downstream Consumers:

  • AccessControlManager for rule evaluation

  • Audit systems for logging policy decisions

  • Policy analysis tools for rule validation

Usage Patterns:

  • Created by administrators to define access policies

  • Evaluated during workflow and node execution

  • Cached for performance optimization

  • Updated when policies change

Implementation Details:

Uses dataclass for efficient serialization and comparison. Supports priority-based rule ordering for conflict resolution. Includes expiration for time-limited permissions. Conditions enable complex policy logic.

Example

>>> rule = PermissionRule(
...     id="allow_analysts_read",
...     resource_type="node",
...     resource_id="sensitive_data",
...     permission=NodePermission.READ_OUTPUT,
...     effect=PermissionEffect.ALLOW,
...     role="analyst"
... )
>>> print(rule.id)
allow_analysts_read
Parameters:
__init__(id: str, resource_type: str, resource_id: str, permission: WorkflowPermission | NodePermission, effect: PermissionEffect, user_id: str | None = None, role: str | None = None, tenant_id: str | None = None, conditions: dict[str, ~typing.Any]=<factory>, created_at: datetime = <factory>, created_by: str | None = None, expires_at: datetime | None = None, priority: int = 0) → None
Parameters:
Return type:

None

created_by: str | None = None
expires_at: datetime | None = None
priority: int = 0
role: str | None = None
tenant_id: str | None = None
user_id: str | None = None
id: str
resource_type: str
resource_id: str
permission: WorkflowPermission | NodePermission
effect: PermissionEffect
conditions: dict[str, Any]
created_at: datetime

Access Decisions

class kailash.access_control.AccessDecision(allowed: bool, reason: str, applied_rules: list[PermissionRule] = <factory>, conditions_met: dict[str, bool]=<factory>, masked_fields: list[str] = <factory>, redirect_node: str | None = None)

Bases: object

Result of an access control decision.

Contains the outcome of evaluating access control rules for a specific user and resource, including whether access is allowed, the reasoning, and any additional actions required (like data masking).

Design Purpose:

Provides a comprehensive result object that captures not just the allow/deny decision, but also the context and reasoning behind it. Enables audit logging, debugging, and conditional execution.

Upstream Dependencies:

  • AccessControlManager rule evaluation

  • Permission rule matching logic

  • Conditional evaluation systems

Downstream Consumers:

  • AccessControlledRuntime for execution decisions

  • Audit logging systems for compliance

  • Error handling for access denial messages

  • Data masking systems for output filtering

Usage Patterns:

  • Returned by all access control check methods

  • Logged for audit and debugging purposes

  • Used to determine execution flow

  • Provides user-friendly error messages

Implementation Details:

Immutable after creation to ensure decision integrity. Includes applied rules for transparency and debugging. Supports conditional decisions for complex scenarios. Contains masking information for data protection.

Example

>>> decision = AccessDecision(
...     allowed=True,
...     reason="User has analyst role",
...     masked_fields=["ssn", "phone"]
... )
>>> print(decision.allowed)
True
Parameters:
__init__(allowed: bool, reason: str, applied_rules: list[PermissionRule] = <factory>, conditions_met: dict[str, bool]=<factory>, masked_fields: list[str] = <factory>, redirect_node: str | None = None) → None
Parameters:
Return type:

None

redirect_node: str | None = None
allowed: bool
reason: str
applied_rules: list[PermissionRule]
conditions_met: dict[str, bool]
masked_fields: list[str]

Enumerations

class kailash.access_control.WorkflowPermission(value)

Bases: Enum

Workflow-level permissions

VIEW = 'view'
EXECUTE = 'execute'
MODIFY = 'modify'
DELETE = 'delete'
SHARE = 'share'
ADMIN = 'admin'
classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name)

Return the member matching name.

classmethod __iter__()

Return members in definition order.

classmethod __len__()

Return the number of members (no aliases)

class kailash.access_control.NodePermission(value)

Bases: Enum

Node-level permissions

EXECUTE = 'execute'
READ_OUTPUT = 'read_output'
WRITE_INPUT = 'write_input'
SKIP = 'skip'
MASK_OUTPUT = 'mask_output'
classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name)

Return the member matching name.

classmethod __iter__()

Return members in definition order.

classmethod __len__()

Return the number of members (no aliases)

class kailash.access_control.PermissionEffect(value)

Bases: Enum

Effect of a permission rule

ALLOW = 'allow'
DENY = 'deny'
CONDITIONAL = 'conditional'
classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name)

Return the member matching name.

classmethod __iter__()

Return members in definition order.

classmethod __len__()

Return the number of members (no aliases)

Access Controlled Runtime

Access-Controlled Runtime for Kailash SDK

This module provides an access-controlled runtime that wraps the standard runtime to add permission checks. The standard runtime remains unchanged, ensuring complete backward compatibility.

Users who don’t need access control continue using LocalRuntime as normal. Users who need access control use AccessControlledRuntime instead.

Example without access control (existing code):
>>> from kailash.runtime.local import LocalRuntime
>>> from kailash.workflow import Workflow
>>> runtime = LocalRuntime()
>>> workflow = Workflow(workflow_id="test", name="Test")
>>> result, run_id = runtime.execute(workflow)  # Works exactly as before
Example with access control (opt-in):
>>> from kailash.runtime.access_controlled import AccessControlledRuntime
>>> from kailash.access_control import UserContext, get_access_control_manager
>>> user = UserContext(user_id="123", tenant_id="abc", email="user@test.com", roles=["analyst"])
>>> runtime = AccessControlledRuntime(user_context=user)
>>> # Access control manager is disabled by default for compatibility
>>> acm = get_access_control_manager()
>>> acm.enabled  # Should be False by default
False
class kailash.runtime.access_controlled.AccessControlledRuntime(user_context: UserContext, base_runtime: LocalRuntime | None = None)[source]

Bases: object

Runtime with transparent access control layer.

This runtime wraps the standard LocalRuntime and adds access control checks without modifying the original runtime or requiring any changes to existing nodes or workflows.

Design Purpose:

Provides a drop-in replacement for LocalRuntime that adds security without breaking existing workflows. Enables role-based access control, data masking, and conditional execution based on user permissions.

Upstream Dependencies:

  • AccessControlManager for permission evaluation

  • UserContext from authentication systems

  • LocalRuntime for actual workflow execution

  • PermissionRule definitions from configuration

Downstream Consumers:

  • Applications requiring secure workflow execution

  • Multi-tenant systems with user isolation

  • Audit systems for compliance logging

  • Data governance systems for access tracking

Usage Patterns:

  • Used as direct replacement for LocalRuntime

  • Configured with user context during initialization

  • Integrates with JWT authentication systems

  • Supports both workflow and node-level permissions

Implementation Details:

Wraps LocalRuntime and intercepts workflow execution to add permission checks. Creates access-controlled node wrappers that evaluate permissions before execution. Supports data masking, conditional routing, and fallback execution.

Error Handling:

  • Access denied raises PermissionError with clear messages

  • Missing permissions default to deny for security

  • Configuration errors are logged and treated as disabled

  • Evaluation errors fall back to base runtime behavior

Side Effects:

  • Logs all access decisions for audit purposes

  • May redirect execution to alternative nodes

  • Applies data masking to sensitive outputs

  • Caches permission decisions for performance

Example

>>> from kailash.runtime.access_controlled import AccessControlledRuntime
>>> from kailash.access_control import UserContext
>>> from kailash.workflow import Workflow
>>>
>>> user = UserContext(user_id="123", tenant_id="abc", email="user@test.com", roles=["analyst"])
>>> runtime = AccessControlledRuntime(user_context=user)
>>> # By default, access control is disabled for backward compatibility
>>> workflow = Workflow(workflow_id="test", name="Test Workflow")
>>> isinstance(runtime, AccessControlledRuntime)
True
Parameters:
__init__(user_context: UserContext, base_runtime: LocalRuntime | None = None)[source]

Initialize access-controlled runtime.

Parameters:

  • user_context (UserContext) – The user context for access control decisions

  • base_runtime (LocalRuntime | None) – The underlying runtime to use (defaults to LocalRuntime)

execute(workflow: Workflow, parameters: dict[str, Any] | None = None, *, soft_time_limit: float | None = None, time_limit: float | None = None, **kwargs: Any) → tuple[Any, str | None][source]

Execute workflow with access control.

This method has the exact same signature as the standard runtime, ensuring complete compatibility.

Parameters:

  • workflow (Workflow) – Workflow to execute.

  • parameters (dict[str, Any] | None) – Optional parameter overrides per node.

  • soft_time_limit (float | None) – Optional advisory deadline in seconds (#912 Shard 1 slot, forwarded to inner runtime).

  • time_limit (float | None) – Optional unconditional kill deadline in seconds.

  • **kwargs (Any) – Forward-compatibility kwargs forwarded to inner runtime.

Return type:

tuple[Any, str | None]

close() → None[source]

Close the runtime and clean up resources.

Only closes the base runtime if it was created by this instance.

Return type:

None

__enter__() → AccessControlledRuntime[source]

Enter context manager.

Return type:

AccessControlledRuntime

__exit__(exc_type, exc_val, exc_tb)[source]

Exit context manager.

class kailash.runtime.access_controlled.AccessControlConfig[source]

Bases: object

Configuration for access control in workflows.

Provides a declarative way to define access rules without modifying workflow code. Enables administrators to configure permissions externally from workflow definitions.

Design Purpose:

Separates access control policy from workflow implementation, enabling dynamic permission changes without code modifications. Supports both workflow-level and node-level permission rules.

Upstream Dependencies:

  • Administrative interfaces for rule creation

  • Configuration management systems

  • Policy definition templates

Downstream Consumers:

  • AccessControlManager for rule application

  • AccessControlledRuntime for secure execution

  • Policy management tools for validation

Usage Patterns:

  • Created by administrators or configuration systems

  • Applied to workflows before execution

  • Used for testing different access scenarios

  • Integrated with external policy management

Implementation Details:

Maintains list of PermissionRule objects with helper methods for adding common rule types. Rules are applied to manager in batch for consistency.

Example

>>> config = AccessControlConfig()
>>> config.add_workflow_permission(
...     workflow_id="analytics",
...     permission=WorkflowPermission.EXECUTE,
...     role="analyst"
... )
>>> config.add_node_permission(
...     workflow_id="analytics",
...     node_id="sensitive_data",
...     permission=NodePermission.READ_OUTPUT,
...     role="admin"
... )
__init__()[source]
add_workflow_permission(workflow_id: str, permission: WorkflowPermission, user_id: str | None = None, role: str | None = None, effect: PermissionEffect = PermissionEffect.ALLOW)[source]

Add a workflow-level permission rule

Parameters:
add_node_permission(workflow_id: str, node_id: str, permission: NodePermission, user_id: str | None = None, role: str | None = None, effect: PermissionEffect = PermissionEffect.ALLOW, masked_fields: list[str] | None = None, redirect_node: str | None = None)[source]

Add a node-level permission rule

Parameters:
apply_to_manager(manager: AccessControlManager)[source]

Apply all rules to an access control manager

Parameters:

manager (AccessControlManager)

kailash.runtime.access_controlled.execute_with_access_control(workflow: Workflow, user_context: UserContext, parameters: dict[str, Any] | None = None, access_config: AccessControlConfig | None = None) → tuple[Any, str | None][source]

Convenience function to execute a workflow with access control.

Provides a simple way to execute workflows with access control without manually creating runtime instances. Automatically applies access configuration and manages the runtime lifecycle.

Parameters:

  • workflow (Workflow) – The workflow to execute

  • user_context (UserContext) – User context for access control decisions

  • parameters (dict[str, Any] | None) – Optional runtime parameters for workflow execution

  • access_config (AccessControlConfig | None) – Optional access control configuration to apply

Returns:

  • result: The workflow execution result

  • run_id: Unique identifier for this execution run

Return type:

Tuple containing

Raises:
Side Effects:

  • Applies access control rules to global manager if config provided

  • Logs audit events for access decisions

  • Enables access control globally during execution

Example

>>> from kailash.runtime.access_controlled import execute_with_access_control
>>> from kailash.access_control import UserContext
>>> from kailash.workflow import Workflow
>>>
>>> user = UserContext(user_id="123", tenant_id="abc", email="user@test.com", roles=["viewer"])
>>> workflow = Workflow(workflow_id="test", name="Test")
>>> # Function exists and can be called
>>> callable(execute_with_access_control)
True

Base Nodes with ACL

Base Node with Optional Access Control Layer

This module extends the base Node class with optional access control capabilities. The access control is completely transparent and disabled by default, ensuring no interference with existing SDK usage.

Key Design Principles: - Access control is OFF by default - Zero performance impact when disabled - Fully backward compatible - Opt-in at workflow or node level - No changes required to existing code

class kailash.nodes.base_with_acl.NodeWithAccessControl(**config)[source]

Bases: Node

Base node class with optional access control capabilities.

Extends the standard Node class with transparent access control features that can be enabled on demand without affecting existing functionality. Access control is completely disabled by default for backward compatibility.

Design Purpose:

Provides a foundation for nodes that need access control while maintaining complete backward compatibility. Enables fine-grained permissions, data masking, and conditional execution.

Upstream Dependencies:

  • AccessControlManager for permission evaluation

  • UserContext from authentication systems

  • PermissionRule definitions from configuration

Downstream Consumers:

  • AccessControlledRuntime for secure execution

  • Audit systems for logging access attempts

  • Data masking systems for output filtering

Usage Patterns:

  • Extended by nodes requiring access control

  • Configured with permission requirements

  • Used in conjunction with AccessControlledRuntime

  • Transparent to existing node implementations

Implementation Details:

Access control is evaluated only when explicitly enabled. Permissions checked before node execution. Output masking applied based on user roles. Fallback execution for denied access scenarios.

Error Handling:

  • Access denied returns user-friendly error messages

  • Missing permissions default to deny

  • Configuration errors are logged and treated as disabled

  • Execution errors maintain standard Node behavior

Side Effects:

  • Logs access attempts for audit purposes

  • May redirect execution to fallback nodes

  • Applies data masking to sensitive outputs

Example

>>> class SecureProcessorNode(NodeWithAccessControl):
...     def _execute(self, **inputs):
...         return {"result": "processed"}
>>>
>>> node = SecureProcessorNode(
...     enable_access_control=True,
...     required_permission=NodePermission.EXECUTE,
...     mask_output_fields=["sensitive_data"]
... )
__init__(**config)[source]

Initialize the node with configuration parameters.

This method performs the following initialization steps:

  1. Sets the node ID (defaults to class name)

  2. Creates metadata from provided arguments

  3. Sets up logging for the node

  4. Stores configuration in self.config

  5. Validates configuration against parameters

The configuration is validated by calling _validate_config(), which checks that all required parameters are present and of the correct type.

Parameters:

**kwargs – Configuration parameters including: - id: Optional custom node ID - name: Display name for the node - description: Node description - version: Node version - author: Node author - tags: Set of tags for discovery - Any parameters defined in get_parameters()

Raises:

NodeConfigurationError – If configuration is invalid or if metadata validation fails

Downstream effects:

  • Creates self.metadata for discovery

  • Sets up self.logger for execution logging

  • Stores self.config for runtime access

  • Validates parameters are correctly specified

run(**inputs) → Any[source]

Execute node with optional access control checks.

If access control is disabled or no user context is present, this behaves exactly like the standard Node.execute() method.

Return type:

Any

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

  1. is not in _INIT_CAPTURE_EXCLUDE,

  2. is not already present in self.config (subclass may have set it directly via **kwargs forwarding),

  3. has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:

None

execute(**runtime_inputs) → dict[str, Any]

Execute the node with validation and error handling.

This is the main entry point for node execution that orchestrates the complete execution lifecycle:

  1. Input validation (validate_inputs)

  2. Execution (run)

  3. Output validation (validate_outputs)

  4. Error handling and logging

  5. Performance metrics

Execution flow:

  1. Logs execution start

  2. Validates inputs against parameter schema

  3. Calls run() with validated inputs

  4. Validates outputs are JSON-serializable

  5. Logs execution time

  6. Returns validated outputs

Error handling strategy:

  • NodeValidationError: Re-raised as-is (input/output issues)

  • NodeExecutionError: Re-raised as-is (run() failures)

  • Other exceptions: Wrapped in NodeExecutionError

Performance tracking:

  • Records execution start/end times

  • Logs total execution duration

  • Includes timing in execution logs

Returns:

Dictionary of validated outputs from run()

Raises:

  • NodeExecutionError – If execution fails in run()

  • NodeValidationError – If input/output validation fails

Return type:

dict[str, Any]

Called by:

  • LocalRuntime: During workflow execution

  • TaskManager: With execution tracking

  • Unit tests: For node testing

Downstream effects:

  • Logs provide execution history

  • Metrics enable performance monitoring

  • Validation ensures data integrity

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

  • enabled: Whether caching is enabled

  • size: Current cache size

  • max_size: Maximum cache size

  • hits: Number of cache hits

  • misses: Number of cache misses

  • evictions: Number of cache evictions

  • hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

  1. Type validation during execution

  2. Documentation for downstream consumers

  3. Workflow connection validation

  4. Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }
Returns:

Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)

Return type:

dict[str, NodeParameter]

Used by:

  • validate_outputs(): Validates runtime outputs

  • Workflow.connect(): Validates connections between nodes

  • Documentation generators: Create output documentation

  • Export systems: Include output schemas in manifests

abstractmethod get_parameters() → dict[str, NodeParameter]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

  1. What inputs the node expects

  2. Type requirements for each input

  3. Whether inputs are required or optional

  4. Default values for optional inputs

  5. Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

  • During initialization: _validate_config() checks configuration

  • During execution: validate_inputs() validates runtime data

  • During workflow creation: Used for connection validation

  • During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }
Returns:

Dictionary mapping parameter names to their definitions

Return type:

dict[str, NodeParameter]

Used by:

  • _validate_config(): Validates configuration matches parameters

  • validate_inputs(): Validates runtime inputs

  • to_dict(): Includes parameters in serialization

  • Workflow.connect(): Validates compatible connections

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

  • key (str) – The key to retrieve from the workflow context

  • default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)
property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:

NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

  • key (str) – The key to store the value under

  • value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)
to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

  1. Workflow export

  2. Node persistence

  3. API responses

  4. Configuration sharing

The serialized format includes:

  • id: Unique node identifier

  • type: Node class name

  • metadata: Complete node metadata

  • config: Current configuration

  • parameters: Parameter definitions with types

Type serialization:

  • Python types are converted to string names

  • Complex types may require custom handling

  • Parameter defaults are included

Returns:

  • Node identification and type

  • Complete metadata

  • Configuration values

  • Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

  • WorkflowExporter: For workflow serialization

  • CLI: For node inspection

  • API: For node information endpoints

  • Debugging: For node state inspection

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

  1. Gets parameter definitions from get_parameters()

  2. Checks each parameter for:

    • Presence (if required)

    • Type compatibility

    • Null handling for optional parameters

  3. Attempts type conversion if needed

  4. Applies default values for missing optional parameters

Key behaviors:

  • Required parameters must be provided or have defaults

  • Optional parameters can be None

  • Type mismatches attempt conversion before failing

  • Error messages include parameter descriptions

Example flow:

# Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

  • Type conversions applied

  • Defaults for missing optional parameters

  • None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

  • execute(): Before passing inputs to run()

  • Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

  1. Schema validation: If get_output_schema() is defined, validates types and required fields

  2. JSON serialization: Ensures all outputs can be serialized

Validation process:

  1. Check outputs is a dictionary

  2. If output schema exists:

    • Validate required fields are present

    • Check type compatibility

    • Attempt type conversion if needed

  3. Verify JSON-serializability

  4. Return validated outputs

Schema validation features:

  • Required outputs must be present

  • Optional outputs can be None or missing

  • Type mismatches attempt conversion

  • Clear error messages with field details

Parameters:

outputs (dict[str, Any]) – Outputs to validate from run() method

Returns:

The same outputs dictionary if valid

Raises:

NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values

Return type:

dict[str, Any]

Called by:

  • execute(): After run() completes

  • Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:

patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache

Return type:

None

class kailash.nodes.base_with_acl.AsyncNodeWithAccessControl(**config)[source]

Bases: AsyncNode

Async version of NodeWithAccessControl

__init__(**config)[source]

Initialize AsyncNode with all enterprise capabilities.

This calls the MRO chain to initialize all mixins and the base Node. The MRO ensures each mixin’s __init__ is called exactly once.

Parameters:

**kwargs – Configuration parameters for node and mixins - All Node parameters (node_id, node_type, config, etc.) - security_config: Optional SecurityConfig for SecurityMixin - log_level: Log level for LoggingMixin (default: “INFO”) - enable_performance_tracking: Enable performance metrics (default: True)

async async_run(**inputs) → Any[source]

Async execution with optional access control

Return type:

Any

classmethod __init_subclass__(**subclass_kwargs)

Install a per-subclass __init__ wrapper that captures bound init params.

Issue #929: Workflow.to_dict() → Workflow.from_dict() silently strips every named/positional argument that a subclass __init__ consumes WITHOUT re-injecting into self.config. PythonCodeNode.__init__ consumes code, input_types, output_type, etc. as named args; none of them flow into super().__init__(**kwargs) because they were peeled off the kwargs dict before the super call.

The fix is applied here ONCE per subclass: wrap cls.__init__ so that AFTER the original init runs (and self.config is populated by Node.__init__), the bound init parameters are merged into self.config for every name that:

  1. is not in _INIT_CAPTURE_EXCLUDE,

  2. is not already present in self.config (subclass may have set it directly via **kwargs forwarding),

  3. has a non-sentinel value (positional defaults pass through; the sentinel for “user passed this” is “binding succeeded” — we keep the bound value verbatim, including None, so round-trip is faithful).

The wrapper is installed exactly once per subclass tree leaf via the _init_capture_installed marker, so re-imports / multiple subclass definitions of the same class do not re-wrap.

Round-trip contract: cls(**self.config) after to_dict/from_dict reconstructs an equivalent node, EXCEPT for params whose values are non-JSON-serializable runtime objects (callables, classes, file handles). Those are still captured into self.config (so the dict carries them in-memory), but Workflow.to_json() will skip or fail on them — that is a separate concern and matches existing behavior.

async audit_log(action: str, details: Dict[str, Any]) → None

Log an audit event (async override).

Overrides SecurityMixin.audit_log to prevent blocking the event loop. Uses asyncio.to_thread() to offload print() to thread pool.

Parameters:

  • action (str) – Action being audited

  • details (Dict[str, Any]) – Additional details about the action

Return type:

None

clear_cache() → None

Clear the parameter resolution cache and reset statistics.

Return type:

None

async emit_node_completed(outputs: Dict[str, Any] | None = None, execution_time_ms: float | None = None)

Emit node completed event.

Parameters:
async emit_node_failed(error: str)

Emit node failed event.

Parameters:

error (str)

async emit_node_progress(progress_percent: float, message: str | None = None)

Emit node progress event.

Parameters:

  • progress_percent (float)

  • message (str | None)

async emit_node_started(inputs: Dict[str, Any] | None = None)

Emit node started event.

Parameters:

inputs (Dict[str, Any] | None)

execute(**runtime_inputs) → dict[str, Any]

Execute the node synchronously by running async code with proper event loop handling.

This enhanced implementation handles all event loop scenarios: 1. No event loop: Create new one with asyncio.run() 2. Event loop running: Use ThreadPoolExecutor with isolated loop 3. Threaded contexts: Proper thread-safe execution 4. Windows compatibility: ProactorEventLoopPolicy support

Parameters:

**runtime_inputs – Runtime inputs for node execution

Returns:

Dictionary of validated outputs

Raises:

  • NodeValidationError – If inputs or outputs are invalid

  • NodeExecutionError – If execution fails

Return type:

dict[str, Any]

async execute_async(**runtime_inputs) → dict[str, Any]

Execute the node asynchronously with validation and error handling.

This method follows the same pattern as execute() but supports asynchronous execution. It performs:

  1. Input validation

  2. Execution via async_run()

  3. Output validation

  4. Error handling and logging

Parameters:

**runtime_inputs – Runtime inputs for node execution

Returns:

Dictionary of validated outputs

Raises:

  • NodeValidationError – If inputs or outputs are invalid

  • NodeExecutionError – If execution fails

Return type:

dict[str, Any]

get_cache_stats() → dict[str, Any]

Get parameter cache statistics.

Returns:

  • enabled: Whether caching is enabled

  • size: Current cache size

  • max_size: Maximum cache size

  • hits: Number of cache hits

  • misses: Number of cache misses

  • evictions: Number of cache evictions

  • hit_rate: Cache hit rate (0-1)

Return type:

Dictionary containing cache statistics

get_output_schema() → dict[str, NodeParameter]

Define output parameters for this node.

This optional method allows nodes to specify their output schema for validation. If not overridden, outputs will only be validated for JSON-serializability.

Design purpose: - Enables static analysis of node outputs - Provides runtime validation of output types - Supports automatic documentation of outputs - Facilitates workflow validation and type checking

The output schema serves similar purposes as input parameters:

  1. Type validation during execution

  2. Documentation for downstream consumers

  3. Workflow connection validation

  4. Export manifest generation

Example

>>> def get_output_schema(self):
...     return {
...         'dataframe': NodeParameter(
...             name='dataframe',
...             type=dict,
...             required=True,
...             description='Processed data as dictionary'
...         ),
...         'row_count': NodeParameter(
...             name='row_count',
...             type=int,
...             required=True,
...             description='Number of rows processed'
...         ),
...         'processing_time': NodeParameter(
...             name='processing_time',
...             type=float,
...             required=False,
...             description='Time taken to process in seconds'
...         )
...     }
Returns:

Dictionary mapping output names to their parameter definitions Empty dict by default (no schema validation)

Return type:

dict[str, NodeParameter]

Used by:

  • validate_outputs(): Validates runtime outputs

  • Workflow.connect(): Validates connections between nodes

  • Documentation generators: Create output documentation

  • Export systems: Include output schemas in manifests

abstractmethod get_parameters() → dict[str, NodeParameter]

Define the parameters this node accepts.

This abstract method must be implemented by all concrete nodes to specify their input schema. The parameters define:

  1. What inputs the node expects

  2. Type requirements for each input

  3. Whether inputs are required or optional

  4. Default values for optional inputs

  5. Documentation for each parameter

The returned dictionary is used throughout the node lifecycle:

  • During initialization: _validate_config() checks configuration

  • During execution: validate_inputs() validates runtime data

  • During workflow creation: Used for connection validation

  • During export: Included in workflow manifests

Example

>>> def get_parameters(self):
...     return {
...         'input_file': NodeParameter(
...             name='input_file',
...             type=str,
...             required=True,
...             description='Path to input CSV file'
...         ),
...         'delimiter': NodeParameter(
...             name='delimiter',
...             type=str,
...             required=False,
...             default=',',
...             description='CSV delimiter character'
...         )
...     }
Returns:

Dictionary mapping parameter names to their definitions

Return type:

dict[str, NodeParameter]

Used by:

  • _validate_config(): Validates configuration matches parameters

  • validate_inputs(): Validates runtime inputs

  • to_dict(): Includes parameters in serialization

  • Workflow.connect(): Validates compatible connections

get_performance_metrics() → list

Get collected performance metrics.

Return type:

list

get_security_context() → Dict[str, Any]

Get current security context.

Return type:

Dict[str, Any]

get_workflow_context(key: str, default: Any | None = None) → Any

Get a value from the workflow context.

This method allows nodes to retrieve shared state from the workflow execution context. The workflow context is managed by the runtime and provides a way for nodes to share data within a single workflow execution.

Parameters:

  • key (str) – The key to retrieve from the workflow context

  • default (Any | None) – Default value to return if key is not found

Returns:

The value from the workflow context, or default if not found

Return type:

Any

Example

>>> # In a transaction node
>>> connection = self.get_workflow_context('transaction_connection')
>>> if connection:
>>>     # Use the shared connection
>>>     result = await connection.execute(query)
has_event_stream() → bool

Check if event stream is available.

Return type:

bool

property id: str

Backward compatibility property for node identifier.

Returns the node’s identifier (_node_id). This property maintains backward compatibility for code that accesses node.id.

The internal identifier is now _node_id to prevent namespace collision with user’s ‘id’ parameter.

async log_error(message: str, error: Exception | None = None, **extra) → None

Log error message with context (async override).

Overrides LoggingMixin.log_error to prevent blocking.

Parameters:

  • message (str) – Log message

  • error (Exception | None) – Optional exception to include

  • **extra – Additional context

Return type:

None

async log_error_with_traceback(error: Exception, operation: str = 'unknown') → None

Log an error with full traceback information (async override).

Overrides LoggingMixin.log_error_with_traceback to prevent blocking.

Parameters:

  • error (Exception) – Exception that occurred

  • operation (str) – Operation that failed

Return type:

None

async log_info(message: str, **extra) → None

Log info message with context (async override).

Overrides LoggingMixin.log_info to prevent blocking.

Parameters:

  • message (str) – Log message

  • **extra – Additional context

Return type:

None

async log_node_execution(operation: str, **context) → None

Log node execution information (async override).

Overrides LoggingMixin.log_node_execution to prevent blocking.

Parameters:

  • operation (str) – Type of operation being performed

  • **context – Additional context

Return type:

None

async log_security_event(event: str, level: str = 'INFO') → None

Log a security-related event (async override).

This method provides async logging for security events when audit logging is enabled in security_config.

Parameters:

  • event (str) – Description of the security event

  • level (str) – Log level (INFO, WARNING, ERROR)

Return type:

None

async log_warning(message: str, **extra) → None

Log warning message with context (async override).

Overrides LoggingMixin.log_warning to prevent blocking.

Parameters:

  • message (str) – Log message

  • **extra – Additional context

Return type:

None

async log_with_context(level: str, message: str, **context) → None

Log a message with additional context (async override).

Overrides LoggingMixin.log_with_context to prevent blocking.

Parameters:

  • level (str) – Log level (debug, info, warning, error, critical)

  • message (str) – Log message

  • **context – Additional context to include

Return type:

None

property metadata: NodeMetadata

Backward compatibility property for node metadata.

Returns the node’s internal NodeMetadata object (_node_metadata). This property maintains backward compatibility for code that accesses node.metadata.

The internal metadata is now _node_metadata to prevent namespace collision with user’s ‘metadata’ parameter.

Returns:

NodeMetadata object containing node identification and documentation

Note

Users can now have parameters named “metadata” without conflicts. The parameter will be in node.config[‘metadata’], while this property returns the internal NodeMetadata object.

run(**kwargs) → dict[str, Any]

Synchronous run is not supported for AsyncNode.

AsyncNode subclasses should implement async_run() instead of run(). This method exists to provide a clear error message if someone accidentally tries to implement run() on an async node.

Raises:

NotImplementedError – Always, as async nodes must use async_run()

Return type:

dict[str, Any]

set_event_context(event_stream: EventStream, session_id: str | None = None, workflow_id: str | None = None, execution_id: str | None = None)

Set the event context for this node.

Parameters:

  • event_stream (EventStream)

  • session_id (str | None)

  • workflow_id (str | None)

  • execution_id (str | None)

set_log_context(**context)

Set logging context.

set_security_context(context: Dict[str, Any]) → None

Set security context for the node.

Parameters:

context (Dict[str, Any])

Return type:

None

set_workflow_context(key: str, value: Any) → None

Set a value in the workflow context.

This method allows nodes to store shared state in the workflow execution context. Other nodes in the same workflow execution can retrieve this data using get_workflow_context().

Parameters:

  • key (str) – The key to store the value under

  • value (Any) – The value to store in the workflow context

Return type:

None

Example

>>> # In a transaction scope node
>>> connection = await self.get_connection()
>>> transaction = await connection.begin()
>>> self.set_workflow_context('transaction_connection', connection)
>>> self.set_workflow_context('active_transaction', transaction)
to_dict() → dict[str, Any]

Convert node to dictionary representation.

Serializes the node instance to a dictionary format suitable for:

  1. Workflow export

  2. Node persistence

  3. API responses

  4. Configuration sharing

The serialized format includes:

  • id: Unique node identifier

  • type: Node class name

  • metadata: Complete node metadata

  • config: Current configuration

  • parameters: Parameter definitions with types

Type serialization:

  • Python types are converted to string names

  • Complex types may require custom handling

  • Parameter defaults are included

Returns:

  • Node identification and type

  • Complete metadata

  • Configuration values

  • Parameter schemas

Return type:

Dictionary representation containing

Raises:

NodeExecutionError – If serialization fails due to: - get_parameters() errors - Metadata serialization issues - Type conversion problems

Used by:

  • WorkflowExporter: For workflow serialization

  • CLI: For node inspection

  • API: For node information endpoints

  • Debugging: For node state inspection

track_performance(func)

Decorator to track method performance.

async validate_and_sanitize_inputs(inputs: Dict[str, Any]) → Dict[str, Any]

Validate and sanitize input parameters (async override).

Overrides SecurityMixin.validate_and_sanitize_inputs when the full SecurityMixin from mixins.py is used (with logging).

Parameters:

inputs (Dict[str, Any]) – Dictionary of input parameters

Returns:

Dictionary of validated and sanitized parameters

Return type:

Dict[str, Any]

validate_inputs(**kwargs) → dict[str, Any]

Validate runtime inputs against node requirements.

This method validates inputs provided at execution time against the node’s parameter schema. It ensures type safety and provides helpful error messages for invalid inputs.

Validation steps:

  1. Gets parameter definitions from get_parameters()

  2. Checks each parameter for:

    • Presence (if required)

    • Type compatibility

    • Null handling for optional parameters

  3. Attempts type conversion if needed

  4. Applies default values for missing optional parameters

Key behaviors:

  • Required parameters must be provided or have defaults

  • Optional parameters can be None

  • Type mismatches attempt conversion before failing

  • Error messages include parameter descriptions

Example flow:

# Node expects: {‘count’: int, ‘name’: str (optional)} inputs = {‘count’: ‘42’, ‘name’: None} validated = validate_inputs(**inputs) # Returns: {‘count’: 42} # Converted and None removed

Parameters:

**kwargs – Runtime inputs to validate

Returns:

  • Type conversions applied

  • Defaults for missing optional parameters

  • None values removed for optional parameters

Return type:

Dictionary of validated inputs with

Raises:

NodeValidationError – If inputs are invalid: - Missing required parameters - Type conversion failures - get_parameters() errors

Called by:

  • execute(): Before passing inputs to run()

  • Workflow validation: During connection checks

validate_outputs(outputs: dict[str, Any]) → dict[str, Any]

Validate outputs against schema and JSON-serializability.

This enhanced method validates outputs in two ways:

  1. Schema validation: If get_output_schema() is defined, validates types and required fields

  2. JSON serialization: Ensures all outputs can be serialized

Validation process:

  1. Check outputs is a dictionary

  2. If output schema exists:

    • Validate required fields are present

    • Check type compatibility

    • Attempt type conversion if needed

  3. Verify JSON-serializability

  4. Return validated outputs

Schema validation features:

  • Required outputs must be present

  • Optional outputs can be None or missing

  • Type mismatches attempt conversion

  • Clear error messages with field details

Parameters:

outputs (dict[str, Any]) – Outputs to validate from run() method

Returns:

The same outputs dictionary if valid

Raises:

NodeValidationError – If outputs are invalid: - Not a dictionary - Missing required outputs - Type validation failures - Non-serializable values

Return type:

dict[str, Any]

Called by:

  • execute(): After run() completes

  • Test utilities: For output validation

warm_cache(patterns: list[dict[str, Any]]) → None

Warm the cache with known parameter patterns.

Parameters:

patterns (list[dict[str, Any]]) – List of parameter dictionaries to pre-cache

Return type:

None

kailash.nodes.base_with_acl.make_node_access_controlled(node_class, **acl_config)[source]

Factory function to add access control to any existing node class.

This allows adding access control to nodes without modifying their code:

>>> from kailash.nodes.data.readers import CSVReaderNode
>>> SecureCSVReader = make_node_access_controlled(
...     CSVReaderNode,
...     enable_access_control=True,
...     required_permission=NodePermission.READ_OUTPUT
... )
kailash.nodes.base_with_acl.add_access_control(node_instance, **acl_config)[source]

Add access control to an existing node instance.

This function adds access control attributes to a node instance. For simplicity in this example, we’ll just add the attributes and let the AccessControlledRuntime handle the actual access control.

Parameters:

  • node_instance – The node instance to wrap

  • **acl_config – Access control configuration - enable_access_control: Whether to enable access control (default: True) - required_permission: Permission required to execute the node - node_id: Unique identifier for access control rules - mask_output_fields: List of fields to mask in output for non-admin users - fallback_node: Node ID to execute if access is denied

Returns:

Node instance with access control capabilities

Example

>>> reader = CSVReaderNode(file_path="data.csv")
>>> secure_reader = add_access_control(
...     reader,
...     enable_access_control=True,
...     required_permission=NodePermission.EXECUTE,
...     node_id="secure_csv_reader"
... )

Examples

Basic RBAC Setup

from kailash.access_control import (
    UserContext, PermissionRule, NodePermission,
    WorkflowPermission, PermissionEffect, get_access_control_manager
)
from kailash.runtime.access_controlled import AccessControlledRuntime

# Create user context
user = UserContext(
    user_id="john_doe",
    tenant_id="acme_corp",
    email="john@acme.com",
    roles=["analyst", "viewer"]
)

# Configure access control
acm = get_access_control_manager()
acm.enabled = True

# Add permission rules
acm.add_rule(PermissionRule(
    id="allow_analysts_execute",
    resource_type="workflow",
    resource_id="customer_analytics",
    permission=WorkflowPermission.EXECUTE,
    effect=PermissionEffect.ALLOW,
    role="analyst"
))

# Use secure runtime
runtime = AccessControlledRuntime(user_context=user)
result, run_id = runtime.execute(workflow)

Multi-Tenant Isolation

# Create tenant-specific rules
tenant_rule = PermissionRule(
    id="tenant_isolation",
    resource_type="node",
    resource_id="sensitive_data",
    permission=NodePermission.READ_OUTPUT,
    effect=PermissionEffect.ALLOW,
    tenant_id="acme_corp"  # Only ACME users can access
)

acm.add_rule(tenant_rule)

# Users from other tenants will be denied access
other_user = UserContext(
    user_id="jane_smith",
    tenant_id="other_corp",
    email="jane@other.com",
    roles=["admin"]
)

# This will be denied due to tenant mismatch
runtime = AccessControlledRuntime(user_context=other_user)

Data Masking

from kailash.nodes.base_with_acl import add_access_control
from kailash.nodes.data.readers import CSVReaderNode

# Create secure data reader with field masking
secure_reader = add_access_control(
    CSVReaderNode(file_path="customers.csv"),
    enable_access_control=True,
    required_permission=NodePermission.READ_OUTPUT,
    mask_output_fields=["ssn", "phone"]  # Mask for non-admin users
)

workflow.add_node("secure_data", secure_reader)

Permission-Based Routing

# Different processing based on user permissions
admin_processor = PythonCodeNode.from_function(
    lambda data: {"result": process_all_data(data)},
    name="admin_processor"
)

viewer_processor = PythonCodeNode.from_function(
    lambda data: {"result": process_summary_data(data)},
    name="viewer_processor"
)

# Configure different permissions for each path
acm.add_rule(PermissionRule(
    id="admin_full_access",
    resource_type="node",
    resource_id="admin_processor",
    permission=NodePermission.EXECUTE,
    effect=PermissionEffect.ALLOW,
    role="admin"
))

acm.add_rule(PermissionRule(
    id="viewer_limited_access",
    resource_type="node",
    resource_id="viewer_processor",
    permission=NodePermission.EXECUTE,
    effect=PermissionEffect.ALLOW,
    role="viewer"
))

# Runtime will automatically route based on user permissions
workflow.add_node("admin_path", admin_processor)
workflow.add_node("viewer_path", viewer_processor)