1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

"""
Group-based access control implementation
 
Demonstrates how group membership expands into effective permissions.
"""
 
from dataclasses import dataclass, field
from typing import Set, Dict
 
@dataclass
class Group:
    name: str
    members: Set[str] = field(default_factory=set)
 
@dataclass  
class AccessControlSystem:
    # Subject grouping
    groups: Dict[str, Group] = field(default_factory=dict)
    user_groups: Dict[str, Set[str]] = field(default_factory=dict)  # user -> groups
    
    # ACL storage (group-based)
    acls: Dict[str, Dict[str, Set[str]]] = field(default_factory=dict)
    # object -> { principal -> permissions }
    # principal can be user or group
    
    def add_user_to_group(self, user: str, group_name: str):
        """Add user to a group"""
        if group_name not in self.groups:
            self.groups[group_name] = Group(group_name)
        self.groups[group_name].members.add(user)
        
        if user not in self.user_groups:
            self.user_groups[user] = set()
        self.user_groups[user].add(group_name)
    
    def grant_permission(self, principal: str, obj: str, permission: str):
        """Grant permission to user or group"""
        if obj not in self.acls:
            self.acls[obj] = {}
        if principal not in self.acls[obj]:
            self.acls[obj][principal] = set()
        self.acls[obj][principal].add(permission)
    
    def get_effective_permissions(self, user: str, obj: str) -> Set[str]:
        """
        Calculate effective permissions by combining:
        1. Direct user permissions
        2. Permissions from all groups the user belongs to
        """
        effective = set()
        
        if obj not in self.acls:
            return effective
        
        obj_acl = self.acls[obj]
        
        # Direct user permissions
        if user in obj_acl:
            effective.update(obj_acl[user])
        
        # Group permissions
        user_group_set = self.user_groups.get(user, set())
        for group_name in user_group_set:
            if group_name in obj_acl:
                effective.update(obj_acl[group_name])
        
        return effective
    
    def check_access(self, user: str, obj: str, permission: str) -> bool:
        """Check if user has permission on object"""
        effective = self.get_effective_permissions(user, obj)
        return permission in effective
    
    def get_compression_stats(self) -> dict:
        """Show how groups compress the access matrix"""
        total_users = len(self.user_groups)
        total_objects = len(self.acls)
        total_groups = len(self.groups)
        
        # Without groups: users * objects * avg_permissions
        acl_entries = sum(
            len(principals) 
            for principals in self.acls.values()
        )
        
        # Uncompressed would be: users * objects potential
        potential = total_users * total_objects
        
        return {
            "users": total_users,
            "objects": total_objects,
            "groups": total_groups,
            "acl_entries": acl_entries,
            "potential_entries": potential,
            "compression_ratio": potential / max(acl_entries, 1)
        }
 
 
# Example usage:
acs = AccessControlSystem()
 
# Create groups
for i in range(100):
    acs.add_user_to_group(f"dev_{i}", "developers")
for i in range(50):
    acs.add_user_to_group(f"ops_{i}", "operations")
    
# Grant permissions to groups instead of individuals
for i in range(10000):
    acs.grant_permission("developers", f"code_file_{i}", "read")
    acs.grant_permission("developers", f"code_file_{i}", "write")
    
for i in range(1000):
    acs.grant_permission("operations", f"config_{i}", "read")
    acs.grant_permission("operations", f"config_{i}", "write")
 
# Check access
print(acs.check_access("dev_42", "code_file_123", "read"))  # True
print(acs.check_access("ops_10", "code_file_123", "read"))  # False
 
# Compression stats
stats = acs.get_compression_stats()
print(f"Compression ratio: {stats['compression_ratio']:.1f}x")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

"""
Role-Based Access Control (RBAC) implementation
Following NIST RBAC standard model
"""
 
from dataclasses import dataclass, field
from typing import Set, Dict, Optional
 
@dataclass
class RBAC:
    # Core sets
    users: Set[str] = field(default_factory=set)
    roles: Set[str] = field(default_factory=set)
    permissions: Set[str] = field(default_factory=set)
    
    # Assignments
    user_assignment: Dict[str, Set[str]] = field(default_factory=dict)  # user -> roles
    permission_assignment: Dict[str, Set[str]] = field(default_factory=dict)  # role -> permissions
    
    # Role hierarchy: role -> parent roles (inherits from)
    role_hierarchy: Dict[str, Set[str]] = field(default_factory=dict)
    
    # Constraints
    ssd_constraints: Set[frozenset] = field(default_factory=set)  # sets of mutually exclusive roles
    
    def create_role(self, role: str, parent: Optional[str] = None):
        """Create a role, optionally inheriting from parent"""
        self.roles.add(role)
        if parent:
            if role not in self.role_hierarchy:
                self.role_hierarchy[role] = set()
            self.role_hierarchy[role].add(parent)
    
    def assign_permission_to_role(self, permission: str, role: str):
        """Add permission to role (PA relation)"""
        self.permissions.add(permission)
        if role not in self.permission_assignment:
            self.permission_assignment[role] = set()
        self.permission_assignment[role].add(permission)
    
    def assign_role_to_user(self, role: str, user: str) -> bool:
        """Add role to user (UA relation), checking SSD constraints"""
        self.users.add(user)
        
        if user not in self.user_assignment:
            self.user_assignment[user] = set()
        
        # Check SSD constraints
        new_roles = self.user_assignment[user] | {role}
        for constraint in self.ssd_constraints:
            if len(new_roles & constraint) > 1:
                return False  # Would violate SSD
        
        self.user_assignment[user].add(role)
        return True
    
    def add_ssd_constraint(self, conflicting_roles: Set[str]):
        """Add Static Separation of Duty constraint"""
        self.ssd_constraints.add(frozenset(conflicting_roles))
    
    def get_inherited_roles(self, role: str) -> Set[str]:
        """Get all roles inherited by this role (transitive)"""
        inherited = set()
        to_process = [role]
        
        while to_process:
            current = to_process.pop()
            parents = self.role_hierarchy.get(current, set())
            for parent in parents:
                if parent not in inherited:
                    inherited.add(parent)
                    to_process.append(parent)
        
        return inherited
    
    def get_role_permissions(self, role: str) -> Set[str]:
        """Get all permissions for a role (including inherited)"""
        permissions = set()
        
        # Direct permissions
        if role in self.permission_assignment:
            permissions.update(self.permission_assignment[role])
        
        # Inherited permissions
        for parent in self.get_inherited_roles(role):
            if parent in self.permission_assignment:
                permissions.update(self.permission_assignment[parent])
        
        return permissions
    
    def get_user_permissions(self, user: str) -> Set[str]:
        """Get all permissions for a user (union of all role permissions)"""
        permissions = set()
        
        for role in self.user_assignment.get(user, set()):
            permissions.update(self.get_role_permissions(role))
        
        return permissions
    
    def check_access(self, user: str, permission: str) -> bool:
        """Check if user has permission"""
        return permission in self.get_user_permissions(user)
 
 
# Example: Software company RBAC
rbac = RBAC()
 
# Create role hierarchy
rbac.create_role("employee")
rbac.create_role("developer", parent="employee")
rbac.create_role("senior_developer", parent="developer")
rbac.create_role("tech_lead", parent="senior_developer")
rbac.create_role("auditor", parent="employee")
 
# Assign permissions to roles
rbac.assign_permission_to_role("office:enter", "employee")
rbac.assign_permission_to_role("email:send", "employee")
 
rbac.assign_permission_to_role("code:read", "developer")
rbac.assign_permission_to_role("code:write", "developer")
 
rbac.assign_permission_to_role("code:review", "senior_developer")
rbac.assign_permission_to_role("code:merge", "senior_developer")
 
rbac.assign_permission_to_role("project:manage", "tech_lead")
rbac.assign_permission_to_role("team:assign", "tech_lead")
 
rbac.assign_permission_to_role("audit:access", "auditor")
rbac.assign_permission_to_role("audit:report", "auditor")
 
# SSD constraint: cannot be both developer and auditor
rbac.add_ssd_constraint({"developer", "auditor"})
 
# Assign roles to users
rbac.assign_role_to_user("tech_lead", "alice")
rbac.assign_role_to_user("developer", "bob")
 
# This would fail due to SSD:
result = rbac.assign_role_to_user("auditor", "bob")
print(f"Bob as auditor: {result}")  # False - already developer
 
# Check permissions
print(f"Alice permissions: {rbac.get_user_permissions('alice')}")
# Includes: office:enter, email:send, code:read, code:write, 
#           code:review, code:merge, project:manage, team:assign
 
print(f"Can Alice merge code? {rbac.check_access('alice', 'code:merge')}")  # True
print(f"Can Bob merge code? {rbac.check_access('bob', 'code:merge')}")  # False

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155

"""
Attribute-Based Access Control (ABAC) implementation
 
Demonstrates policy-based access decisions using attributes.
"""
 
from dataclasses import dataclass
from typing import Dict, Any, List, Callable
from datetime import datetime, time
import re
 
@dataclass
class AccessRequest:
    subject: Dict[str, Any]     # Subject attributes
    object: Dict[str, Any]      # Object attributes
    action: str                 # Requested action
    environment: Dict[str, Any] # Environment attributes
 
@dataclass
class PolicyRule:
    name: str
    description: str
    condition: Callable[[AccessRequest], bool]
    effect: str  # "permit" or "deny"
    priority: int = 0
 
class ABAC:
    def __init__(self):
        self.rules: List[PolicyRule] = []
        self.default_effect = "deny"  # Fail-safe default
    
    def add_rule(self, rule: PolicyRule):
        self.rules.append(rule)
        # Higher priority first
        self.rules.sort(key=lambda r: -r.priority)
    
    def evaluate(self, request: AccessRequest) -> tuple[str, str]:
        """
        Evaluate access request against all policies.
        Returns (effect, rule_name) for audit trail.
        """
        for rule in self.rules:
            try:
                if rule.condition(request):
                    return rule.effect, rule.name
            except Exception as e:
                # If condition evaluation fails, skip rule
                continue
        
        return self.default_effect, "default_deny"
    
    def check_access(self, request: AccessRequest) -> bool:
        effect, _ = self.evaluate(request)
        return effect == "permit"
 
 
# Example: Hospital Health Record System
 
abac = ABAC()
 
# Rule 1: Doctors can read their own patients' records
abac.add_rule(PolicyRule(
    name="doctor_own_patients",
    description="Doctors can access records of patients assigned to them",
    condition=lambda r: (
        r.subject.get("role") == "doctor" and
        r.object.get("type") == "patient_record" and
        r.object.get("patient_id") in r.subject.get("assigned_patients", []) and
        r.action in ["read", "write"]
    ),
    effect="permit",
    priority=10
))
 
# Rule 2: Nurses can read any patient record in their ward
abac.add_rule(PolicyRule(
    name="nurse_ward_access",
    description="Nurses can read records of patients in their ward",
    condition=lambda r: (
        r.subject.get("role") == "nurse" and
        r.object.get("type") == "patient_record" and
        r.object.get("ward") == r.subject.get("assigned_ward") and
        r.action == "read"
    ),
    effect="permit",
    priority=10
))
 
# Rule 3: Emergency override during emergencies
abac.add_rule(PolicyRule(
    name="emergency_access",
    description="Any medical staff can access records during declared emergency",
    condition=lambda r: (
        r.subject.get("role") in ["doctor", "nurse", "paramedic"] and
        r.object.get("type") == "patient_record" and
        r.environment.get("emergency_mode") == True and
        r.action == "read"
    ),
    effect="permit",
    priority=100  # High priority - overrides other rules
))
 
# Rule 4: Time-based restriction for contractors
abac.add_rule(PolicyRule(
    name="contractor_hours",
    description="Contractors can only access during business hours",
    condition=lambda r: (
        r.subject.get("employment_type") == "contractor" and
        not (9 <= r.environment.get("hour", 0) <= 17)
    ),
    effect="deny",
    priority=50  # Deny rules often have higher priority
))
 
# Rule 5: Block access from untrusted networks
abac.add_rule(PolicyRule(
    name="network_restriction",
    description="Sensitive records cannot be accessed from public networks",
    condition=lambda r: (
        r.object.get("sensitivity") == "high" and
        r.environment.get("network_zone") == "public"
    ),
    effect="deny",
    priority=100
))
 
 
# Test scenarios
 
# Scenario 1: Doctor accessing own patient
request1 = AccessRequest(
    subject={"id": "dr_smith", "role": "doctor", "assigned_patients": ["P001", "P002"]},
    object={"type": "patient_record", "patient_id": "P001", "sensitivity": "normal"},
    action="read",
    environment={"hour": 14, "network_zone": "internal"}
)
print(f"Scenario 1: {abac.evaluate(request1)}")  # ("permit", "doctor_own_patients")
 
# Scenario 2: Doctor accessing unassigned patient (should fail)
request2 = AccessRequest(
    subject={"id": "dr_smith", "role": "doctor", "assigned_patients": ["P001", "P002"]},
    object={"type": "patient_record", "patient_id": "P999", "sensitivity": "normal"},
    action="read",
    environment={"hour": 14, "network_zone": "internal"}
)
print(f"Scenario 2: {abac.evaluate(request2)}")  # ("deny", "default_deny")
 
# Scenario 3: Emergency mode - anyone can access
request3 = AccessRequest(
    subject={"id": "dr_smith", "role": "doctor", "assigned_patients": ["P001", "P002"]},
    object={"type": "patient_record", "patient_id": "P999", "sensitivity": "normal"},
    action="read",
    environment={"hour": 14, "network_zone": "internal", "emergency_mode": True}
)
print(f"Scenario 3: {abac.evaluate(request3)}")  # ("permit", "emergency_access")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

"""
Policy compaction techniques demonstration
"""
 
import re
from fnmatch import fnmatch
from typing import List, Tuple
 
class CompactPolicyEngine:
    def __init__(self):
        # Wildcard-based rules: (subject_pattern, object_pattern, action, effect)
        self.wildcard_rules: List[Tuple[str, str, str, str]] = []
        
        # Default rules for object types
        self.defaults: dict = {}
        
        # Explicit exceptions
        self.exceptions: List[Tuple[str, str, str, str]] = []
    
    def add_wildcard_rule(self, subject_pattern: str, object_pattern: str, 
                          action: str, effect: str):
        self.wildcard_rules.append((subject_pattern, object_pattern, action, effect))
    
    def add_default(self, object_type: str, default_permissions: dict):
        self.defaults[object_type] = default_permissions
    
    def add_exception(self, subject: str, object_id: str, action: str, effect: str):
        self.exceptions.append((subject, object_id, action, effect))
    
    def check_wildcard(self, pattern: str, value: str) -> bool:
        """Check if value matches pattern (supports * and ?)"""
        return fnmatch(value, pattern)
    
    def evaluate(self, subject: str, object_id: str, action: str,
                 object_type: str = None) -> str:
        """
        Evaluate access with compacted policy.
        Order: Exceptions -> Wildcards -> Defaults -> Deny
        """
        
        # Priority 1: Check explicit exceptions
        for exc_subj, exc_obj, exc_action, effect in self.exceptions:
            if exc_subj == subject and exc_obj == object_id and exc_action == action:
                return effect
        
        # Priority 2: Check wildcard rules
        for subj_pattern, obj_pattern, rule_action, effect in self.wildcard_rules:
            if (self.check_wildcard(subj_pattern, subject) and
                self.check_wildcard(obj_pattern, object_id) and
                rule_action in [action, "*"]):
                return effect
        
        # Priority 3: Check defaults
        if object_type and object_type in self.defaults:
            default = self.defaults[object_type]
            if action in default:
                return default[action]
        
        # Default: deny
        return "deny"
 
 
# Example: Document management system
 
policy = CompactPolicyEngine()
 
# Wildcard rules - compact representation of many specific permissions
# "All engineers can read all code"
policy.add_wildcard_rule("engineer:*", "/code/**", "read", "permit")
# "Senior engineers can also write"
policy.add_wildcard_rule("engineer:senior:*", "/code/**", "write", "permit")
# "All employees can read public docs"
policy.add_wildcard_rule("employee:*", "/public/**", "*", "permit")
# "Admins can do anything"
policy.add_wildcard_rule("admin:*", "/**", "*", "permit")
 
# Defaults by object type
policy.add_default("public_doc", {"read": "permit"})
policy.add_default("private_doc", {"read": "deny", "write": "deny"})
 
# Explicit exceptions (override wildcards)
# "Bob cannot access code (on leave)"
policy.add_exception("engineer:bob", "/code/**", "read", "deny")
# "Alice can access private finance docs (special project)"
policy.add_exception("engineer:alice", "/finance/q4-report", "read", "permit")
 
 
# Test the compacted policy
tests = [
    ("engineer:alice", "/code/main.py", "read"),      # permit (wildcard)
    ("engineer:bob", "/code/main.py", "read"),        # deny (exception)
    ("engineer:junior:carol", "/code/main.py", "write"),  # deny (only seniors)
    ("admin:dave", "/finance/secret", "delete"),      # permit (admin wildcard)
    ("employee:eve", "/public/about.txt", "read"),    # permit (employee + public)
]
 
for subject, obj, action in tests:
    result = policy.evaluate(subject, obj, action)
    print(f"{subject} -> {obj}:{action} = {result}")
 
# Compression analysis:
# - Without wildcards: Would need 1000s of specific rules
# - With wildcards: 5 rules + 2 defaults + 2 exceptions = 9 entries
# - Compression ratio could be 100x-1000x depending on actual subjects/objects