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

/**
 * Access Control List (ACL) data structures and operations
 * 
 * This represents how ACLs are typically implemented in file systems.
 */
 
#include <stdint.h>
#include <stdbool.h>
 
// An ACL Entry (ACE)
typedef struct acl_entry {
    uint32_t tag;           // ACL_USER, ACL_GROUP, ACL_USER_OBJ, etc.
    uint32_t qualifier;     // UID or GID (for ACL_USER or ACL_GROUP)
    uint32_t permissions;   // Bitmask of permissions
} acl_entry_t;
 
// Entry tag types (POSIX ACL)
#define ACL_USER_OBJ    0x01    // File owner
#define ACL_USER        0x02    // Named user
#define ACL_GROUP_OBJ   0x04    // File group
#define ACL_GROUP       0x08    // Named group
#define ACL_MASK        0x10    // Maximum permissions for users/groups
#define ACL_OTHER       0x20    // Everyone else
 
// Full ACL structure
typedef struct acl {
    uint32_t count;         // Number of entries
    acl_entry_t entries[];  // Variable-length array of entries
} acl_t;
 
// File structure includes pointer to ACL
typedef struct inode {
    // ... other inode fields ...
    uid_t owner;
    gid_t group;
    mode_t mode;            // Traditional Unix permissions
    acl_t *acl;             // Extended ACL (may be NULL)
} inode_t;
 
/**
 * Check if subject can perform operation on file
 * 
 * This is the ACL checking algorithm used in Linux.
 */
int acl_check_access(
    inode_t *inode,
    uid_t uid, gid_t gid, gid_t *groups, int ngroups,
    int requested_permission
) {
    // Step 1: Check if owner (always use owner permissions)
    if (uid == inode->owner) {
        if ((inode->mode >> 6) & requested_permission)
            return 0;  // ALLOWED
        return -EACCES;  // DENIED
    }
    
    // Step 2: If ACL exists, use ACL checking
    if (inode->acl != NULL) {
        return acl_extended_check(
            inode->acl, uid, gid, groups, ngroups, 
            requested_permission
        );
    }
    
    // Step 3: Fall back to traditional Unix check
    return unix_permission_check(
        inode->mode, inode->group,
        uid, gid, groups, ngroups,
        requested_permission
    );
}
 
/**
 * POSIX ACL extended checking algorithm
 */
int acl_extended_check(
    acl_t *acl, 
    uid_t uid, gid_t gid, gid_t *groups, int ngroups,
    int requested
) {
    int found_group = 0;
    int group_permissions = 0;
    acl_entry_t *mask = NULL;
    
    // First pass: find relevant entries
    for (int i = 0; i < acl->count; i++) {
        acl_entry_t *e = &acl->entries[i];
        
        switch (e->tag) {
            case ACL_USER:
                // Named user entry
                if (e->qualifier == uid) {
                    // Apply mask and check
                    int effective = e->permissions;
                    if (mask) effective &= mask->permissions;
                    if (effective & requested)
                        return 0;  // ALLOWED
                    return -EACCES;  // DENIED
                }
                break;
                
            case ACL_GROUP:
            case ACL_GROUP_OBJ:
                // Check if user is in this group
                gid_t check_gid = (e->tag == ACL_GROUP_OBJ) ? 
                                   gid : e->qualifier;
                if (user_in_group(uid, gid, groups, ngroups, check_gid)) {
                    found_group = 1;
                    group_permissions |= e->permissions;
                }
                break;
                
            case ACL_MASK:
                mask = e;
                break;
                
            case ACL_OTHER:
                // Will use this if no group match
                break;
        }
    }
    
    // Check accumulated group permissions
    if (found_group) {
        int effective = group_permissions;
        if (mask) effective &= mask->permissions;
        if (effective & requested)
            return 0;  // ALLOWED
        return -EACCES;  // DENIED
    }
    
    // Fall through to ACL_OTHER
    for (int i = 0; i < acl->count; i++) {
        if (acl->entries[i].tag == ACL_OTHER) {
            if (acl->entries[i].permissions & requested)
                return 0;  // ALLOWED
            break;
        }
    }
    
    return -EACCES;  // DENIED
}

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

/**
 * Capability-based access control implementation
 * 
 * Unix file descriptors are a form of capability!
 */
 
#include <stdint.h>
#include <stdbool.h>
 
// A capability (as used in capability-based systems)
typedef struct capability {
    uint64_t object_id;     // Reference to the object
    uint32_t rights;        // Permitted operations
    uint32_t flags;         // CAP_INHERITABLE, CAP_DELEGABLE, etc.
    uint64_t cryptographic_check; // For unforgettability verification
} capability_t;
 
// A process's capability list (like a file descriptor table)
typedef struct c_list {
    uint32_t count;
    uint32_t capacity;
    capability_t *capabilities;
} c_list_t;
 
// Process structure includes capability list
typedef struct process {
    // ... other fields ...
    c_list_t c_list;        // Capabilities this process holds
} process_t;
 
/**
 * Capability-based access check
 * 
 * Instead of "does subject have right on object?",
 * we ask "does this capability grant this right?"
 */
int capability_check(
    capability_t *cap,
    uint64_t target_object,
    uint32_t requested_permission
) {
    // Verify capability refers to correct object
    if (cap->object_id != target_object) {
        return -ENOENT;  // Wrong object
    }
    
    // Verify capability grants requested permission
    if ((cap->rights & requested_permission) != requested_permission) {
        return -EACCES;  // Insufficient rights
    }
    
    // Optionally verify cryptographic integrity
    if (!verify_capability_signature(cap)) {
        return -EINVAL;  // Forged or corrupted capability
    }
    
    return 0;  // Access granted
}
 
/**
 * File descriptors as capabilities: Unix open() and access
 */
 
// open() creates a capability (file descriptor)
int sys_open(const char *path, int flags) {
    process_t *proc = current_process();
    
    // Resolve path to inode
    inode_t *inode = namei(path);
    if (!inode) return -ENOENT;
    
    // Check if we can create this capability (ACL check at open time!)
    int access_mode = flags_to_access_mode(flags);
    if (check_permission(inode, proc->euid, proc->egid, access_mode) != 0) {
        return -EACCES;  // Cannot create capability
    }
    
    // Create file object and capability
    file_t *file = create_file(inode, flags);
    
    capability_t cap = {
        .object_id = (uint64_t)file,
        .rights = access_mode_to_rights(access_mode),
        .flags = 0,
    };
    
    // Add to process's capability list and return index (fd)
    int fd = clist_add(&proc->c_list, cap);
    return fd;
}
 
// read() uses the capability (fd) — no further permission check!
ssize_t sys_read(int fd, void *buf, size_t count) {
    process_t *proc = current_process();
    
    // Get capability from process's C-list
    capability_t *cap = clist_get(&proc->c_list, fd);
    if (!cap) return -EBADF;  // No such capability
    
    // Check capability grants read right
    if (!(cap->rights & READ_RIGHT)) {
        return -EBADF;  // Capability doesn't grant read
    }
    
    // Perform read — no ACL check needed!
    file_t *file = (file_t *)cap->object_id;
    return do_read(file, buf, count);
}
 
/**
 * Key insight: Unix file descriptors ARE capabilities!
 * 
 * - open() checks permissions and creates capability (fd)
 * - read()/write() only check capability, not ACL
 * - Capabilities can be delegated (via fork(), sendmsg())
 * - Capabilities can be restricted (fcntl() to reduce rights)
 * - Revocation is tricky (closing fd doesn't revoke delegated copies)
 */

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

/**
 * Capsicum capability-based sandboxing example
 * 
 * Combines ACL-based file opening with pure capability mode
 */
 
#include <sys/capsicum.h>
#include <unistd.h>
#include <fcntl.h>
 
int main(int argc, char *argv[]) {
    // Phase 1: ACL-based authorization
    // Open all files we'll need while we have ambient authority
    
    int input_fd = open("/data/input.txt", O_RDONLY);
    if (input_fd < 0) {
        perror("Cannot open input");
        return 1;
    }
    
    int output_fd = open("/data/output.txt", O_WRONLY | O_CREAT, 0644);
    if (output_fd < 0) {
        perror("Cannot open output");
        return 1;
    }
    
    int log_fd = open("/var/log/myapp.log", O_WRONLY | O_APPEND);
    // log_fd might fail — that's okay
    
    // Phase 2: Limit capabilities on file descriptors
    // Reduce rights to minimum necessary
    
    cap_rights_t rights_read;
    cap_rights_init(&rights_read, CAP_READ, CAP_SEEK);
    cap_rights_limit(input_fd, &rights_read);
    
    cap_rights_t rights_write;
    cap_rights_init(&rights_write, CAP_WRITE);  // No seek, no read
    cap_rights_limit(output_fd, &rights_write);
    
    // Phase 3: Enter capability mode
    // After this, we can ONLY use our existing capabilities (fds)
    
    if (cap_enter() < 0) {
        perror("Cannot enter capability mode");
        return 1;
    }
    
    // NOW WE ARE SANDBOXED
    // - Cannot open() any new files
    // - Cannot access network (no socket fds)
    // - Cannot exec() other programs
    // - Cannot access filesystem via paths
    // - Can ONLY use: input_fd (read/seek), output_fd (write)
    
    // Even if this code has vulnerabilities, attacker is contained
    
    // Phase 4: Process data using only our capabilities
    char buffer[4096];
    ssize_t n;
    
    while ((n = read(input_fd, buffer, sizeof(buffer))) > 0) {
        // Process buffer (potentially vulnerable code)
        process_data(buffer, n);
        
        // Write to output
        write(output_fd, buffer, n);
    }
    
    // Attempting to break out of sandbox will fail:
    // open("/etc/passwd", O_RDONLY);  // Returns ECAPMODE
    // socket(AF_INET, SOCK_STREAM, 0);  // Returns ECAPMODE
    
    return 0;
}
 
/**
 * Capsicum demonstrates the power of hybrid approach:
 * 
 * - ACLs control initial access to files
 * - Capabilities (fds) carry authority into sandbox
 * - Capability mode removes ambient authority
 * - Result: powerful sandboxing with minimal API changes
 */

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

/**
 * Efficient ACL implementation strategies
 */
 
// Strategy 1: Sorted array for binary search
// Good for small to medium ACLs (< 100 entries)
typedef struct {
    uint32_t count;
    struct {
        uint32_t subject_id;    // Sorted by this field
        uint32_t permissions;
    } entries[];
} sorted_acl_t;
 
// O(log n) lookup
int sorted_acl_lookup(sorted_acl_t *acl, uint32_t subject_id) {
    int lo = 0, hi = acl->count - 1;
    while (lo <= hi) {
        int mid = (lo + hi) / 2;
        if (acl->entries[mid].subject_id == subject_id) {
            return acl->entries[mid].permissions;
        } else if (acl->entries[mid].subject_id < subject_id) {
            lo = mid + 1;
        } else {
            hi = mid - 1;
        }
    }
    return 0;  // Not found
}
 
 
// Strategy 2: Hash table for large ACLs
// Good for very large ACLs (1000+ entries)
typedef struct {
    uint32_t bucket_count;
    uint32_t entry_count;
    struct acl_entry {
        uint32_t subject_id;
        uint32_t permissions;
        struct acl_entry *next;
    } **buckets;
} hash_acl_t;
 
// O(1) average lookup
int hash_acl_lookup(hash_acl_t *acl, uint32_t subject_id) {
    uint32_t bucket = subject_id % acl->bucket_count;
    struct acl_entry *e = acl->buckets[bucket];
    while (e) {
        if (e->subject_id == subject_id) {
            return e->permissions;
        }
        e = e->next;
    }
    return 0;
}
 
 
// Strategy 3: Cached/memoized results
// For frequently accessed objects with expensive ACL checks
typedef struct {
    uint64_t subject_id;
    uint64_t object_id;
    uint32_t permissions;
    time_t cached_at;
    time_t ttl;
} acl_cache_entry_t;
 
#define ACL_CACHE_SIZE 1024
acl_cache_entry_t acl_cache[ACL_CACHE_SIZE];
 
int cached_acl_check(uint64_t subject, uint64_t object, uint32_t requested) {
    // Compute cache key
    uint32_t key = (subject ^ object) % ACL_CACHE_SIZE;
    acl_cache_entry_t *e = &acl_cache[key];
    
    // Check for cache hit
    if (e->subject_id == subject && 
        e->object_id == object &&
        time(NULL) - e->cached_at < e->ttl) {
        // Cache hit!
        return (e->permissions & requested) == requested ? 0 : -EACCES;
    }
    
    // Cache miss - perform full ACL check
    int permissions = full_acl_check(subject, object);
    
    // Update cache
    e->subject_id = subject;
    e->object_id = object;
    e->permissions = permissions;
    e->cached_at = time(NULL);
    e->ttl = 60;  // 60 second TTL
    
    return (permissions & requested) == requested ? 0 : -EACCES;
}

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

"""
JWT-based distributed authorization example
 
This demonstrates how tokens carry authorization information
across network boundaries.
"""
 
import jwt
import time
from datetime import datetime, timedelta
 
# Authorization server's secret (in practice, use asymmetric keys)
AUTH_SERVER_SECRET = "super-secret-key-dont-use-in-production"
 
def issue_access_token(user_id: str, groups: list, permissions: dict) -> str:
    """
    Authorization server issues token after authentication
    
    Token contains:
    - Subject identity
    - Group memberships  
    - Granted permissions (capability-like)
    - Expiration time
    """
    now = datetime.utcnow()
    
    payload = {
        "sub": user_id,                        # Subject
        "iat": now,                            # Issued at
        "exp": now + timedelta(hours=1),       # Expires in 1 hour
        "groups": groups,                      # Group memberships
        "permissions": permissions,            # Object:rights mappings
        "iss": "auth.example.com",             # Issuer
    }
    
    token = jwt.encode(payload, AUTH_SERVER_SECRET, algorithm="HS256")
    return token
 
def validate_and_check_access(
    token: str, 
    requested_object: str, 
    requested_permission: str
) -> tuple[bool, str]:
    """
    Resource server validates token and checks authorization
    
    This is the distributed equivalent of checking ACL + capability
    """
    try:
        # Step 1: Validate token signature and expiration
        payload = jwt.decode(
            token, 
            AUTH_SERVER_SECRET, 
            algorithms=["HS256"]
        )
        
        # Step 2: Extract subject identity (for logging/audit)
        subject = payload["sub"]
        groups = payload.get("groups", [])
        
        # Step 3: Check permissions in token (capability-style)
        permissions = payload.get("permissions", {})
        
        if requested_object in permissions:
            allowed = permissions[requested_object]
            if requested_permission in allowed:
                return True, f"Access granted to {subject}"
        
        # Step 4: Check group-based permissions (ACL-style)
        # Would typically lookup object's ACL and check against groups
        # Simplified here:
        if "admin" in groups:
            return True, f"Admin access for {subject}"
        
        return False, f"Access denied for {subject}"
        
    except jwt.ExpiredSignatureError:
        return False, "Token expired"
    except jwt.InvalidTokenError as e:
        return False, f"Invalid token: {e}"
 
# Example usage:
# 1. User authenticates, gets token with embedded permissions
token = issue_access_token(
    user_id="alice@example.com",
    groups=["developers", "project-x"],
    permissions={
        "file:project-x/data.csv": ["read", "write"],
        "file:shared/docs/*": ["read"],
    }
)
 
print(f"Token: {token[:50]}...")
 
# 2. User presents token to resource server
allowed, reason = validate_and_check_access(
    token, 
    "file:project-x/data.csv", 
    "read"
)
print(f"Access to data.csv: {allowed} - {reason}")
 
# 3. Token carries authorization - no backend lookup needed!