Firecracker Security: Multi-Tenant Isolation and Defense-in-Depth Architecture

Table of Contents#

Introduction#

In multi-tenant serverless environments, security is paramount. Firecracker was architected from the ground up with a security-first approach, providing strong isolation guarantees while maintaining minimal overhead. This guide explores Firecracker’s comprehensive security model, from hardware virtualization to process isolation and resource controls.

Firecracker’s security architecture embodies the principle of defense-in-depth, layering multiple security mechanisms to create a robust barrier against potential threats. Understanding these mechanisms is crucial for deploying Firecracker in production environments where customer workloads must be isolated from each other and the host system.

Security Architecture Overview#

1
graph TB
2
    subgraph "Host System Security Layers"
3
        subgraph "Hardware Layer"
4
            VTX[Intel VT-x / AMD-V]
5
            EPT[Extended Page Tables]
6
            SMEP[SMEP/SMAP]
7
        end
8

9
        subgraph "Hypervisor Layer"
10
            KVM[Linux KVM]
11
            IOMMU[IOMMU Isolation]
12
        end
13

14
        subgraph "Process Isolation"
15
            JAI[Jailer Process]
16
            CG[cgroups v2]
17
            NS[Namespaces]
18
            SEC[seccomp-bpf]
19
        end
20

21
        subgraph "Application Security"
22
            FC[Firecracker VMM]
23
            API[REST API]
24
            RATE[Rate Limiting]
25
        end
26

27
        subgraph "Guest Isolation"
28
            VM1[Guest VM 1]
29
            VM2[Guest VM 2]
30
            VM3[Guest VM N]
31
        end
32
    end
33

34
    VTX --> KVM
35
    EPT --> KVM
36
    SMEP --> KVM
37
    KVM --> JAI
38
    IOMMU --> JAI
39
    JAI --> FC
40
    CG --> JAI
41
    NS --> JAI
42
    SEC --> JAI
43
    FC --> API
44
    FC --> RATE
45
    API --> VM1
46
    API --> VM2
47
    API --> VM3

Core Security Principles#

Firecracker’s security design follows these fundamental principles:

Minimal Attack Surface: Firecracker exposes only 5 emulated devices, significantly reducing potential attack vectors Memory Safety: Written in Rust, eliminating entire classes of memory safety vulnerabilities Process Isolation: Each microVM runs in a separate process with strict resource controls Hardware Virtualization: Leverages CPU virtualization features for strong isolation Principle of Least Privilege: Minimal required permissions and system call access

Hardware-Level Security#

CPU Virtualization Features#

1
# Check for hardware virtualization support
2
grep -E 'vmx|svm' /proc/cpuinfo
3

4
# Intel VT-x features
5
grep -E 'vmx|ept|vpid' /proc/cpuinfo
6

7
# AMD-V features
8
grep -E 'svm|npt|nrip' /proc/cpuinfo
9

10
# Check for additional security features
11
grep -E 'smep|smap|pku|cet' /proc/cpuinfo

Memory Protection#

1
#!/usr/bin/env python3
2
import subprocess
3
import re
4

5
def check_memory_security_features():
6
    """Check available memory security features"""
7

8
    features = {}
9

10
    # Check for SMEP (Supervisor Mode Execution Prevention)
11
    try:
12
        result = subprocess.run(['grep', 'smep', '/proc/cpuinfo'],
13
                               capture_output=True, text=True)
14
        features['smep'] = 'smep' in result.stdout
15
    except:
16
        features['smep'] = False
17

18
    # Check for SMAP (Supervisor Mode Access Prevention)
19
    try:
20
        result = subprocess.run(['grep', 'smap', '/proc/cpuinfo'],
21
                               capture_output=True, text=True)
22
        features['smap'] = 'smap' in result.stdout
23
    except:
24
        features['smap'] = False
25

26
    # Check for Intel MPX (Memory Protection Extensions)
27
    try:
28
        result = subprocess.run(['grep', 'mpx', '/proc/cpuinfo'],
29
                               capture_output=True, text=True)
30
        features['mpx'] = 'mpx' in result.stdout
31
    except:
32
        features['mpx'] = False
33

34
    # Check for Intel CET (Control-flow Enforcement Technology)
35
    try:
36
        result = subprocess.run(['grep', 'cet', '/proc/cpuinfo'],
37
                               capture_output=True, text=True)
38
        features['cet'] = 'cet' in result.stdout
39
    except:
40
        features['cet'] = False
41

42
    # Check kernel KASLR (Kernel Address Space Layout Randomization)
43
    try:
44
        with open('/proc/cmdline', 'r') as f:
45
            cmdline = f.read()
46
            features['kaslr'] = 'nokaslr' not in cmdline
47
    except:
48
        features['kaslr'] = False
49

50
    return features
51

52
# Check memory protection
53
features = check_memory_security_features()
54
print("Memory Security Features:")
55
for feature, available in features.items():
56
    status = "✓" if available else "✗"
57
    print(f"  {status} {feature.upper()}")

IOMMU Configuration#

1
# Enable IOMMU for additional isolation
2
# Intel IOMMU
3
echo 'GRUB_CMDLINE_LINUX_DEFAULT="intel_iommu=on iommu=pt"' >> /etc/default/grub
4

5
# AMD IOMMU
6
echo 'GRUB_CMDLINE_LINUX_DEFAULT="amd_iommu=on iommu=pt"' >> /etc/default/grub
7

8
# Update GRUB configuration
9
update-grub
10
reboot
11

12
# Verify IOMMU is enabled
13
dmesg | grep -i iommu
14
ls /sys/kernel/iommu_groups/

Jailer: Process Isolation#

The Jailer is Firecracker’s security wrapper that provides comprehensive process isolation using Linux security primitives.

Basic Jailer Usage#

1
#!/bin/bash
2

3
# Create jailer workspace
4
sudo mkdir -p /srv/jailer
5
sudo chown -R 1000:1000 /srv/jailer
6

7
# Run Firecracker with jailer
8
sudo jailer \
9
  --id vm001 \
10
  --exec-file /usr/local/bin/firecracker \
11
  --uid 1000 \
12
  --gid 1000 \
13
  --chroot-base-dir /srv/jailer \
14
  --netns /proc/self/ns/net \
15
  --daemonize \
16
  --new-pid-ns \
17
  -- \
18
  --api-sock socket \
19
  --config-file vm_config.json

Advanced Jailer Configuration#

1
#!/usr/bin/env python3
2
import json
3
import os
4
import subprocess
5
import tempfile
6
from pathlib import Path
7

8
class SecureJailer:
9
    def __init__(self, base_dir="/srv/jailer"):
10
        self.base_dir = Path(base_dir)
11
        self.instances = {}
12

13
    def create_secure_environment(self, vm_id, config):
14
        """Create a secure isolated environment for a microVM"""
15

16
        # Create VM-specific directories
17
        vm_dir = self.base_dir / vm_id
18
        vm_dir.mkdir(parents=True, exist_ok=True)
19

20
        # Create chroot structure
21
        chroot_dirs = ['bin', 'lib', 'lib64', 'dev', 'proc', 'sys', 'tmp']
22
        for dir_name in chroot_dirs:
23
            (vm_dir / dir_name).mkdir(exist_ok=True)
24

25
        # Copy required binaries and libraries
26
        self._setup_chroot_binaries(vm_dir)
27

28
        # Set up device nodes
29
        self._create_device_nodes(vm_dir / 'dev')
30

31
        # Configure resource limits
32
        cgroup_config = self._setup_cgroups(vm_id, config.get('resources', {}))
33

34
        # Generate jailer command
35
        jailer_cmd = self._build_jailer_command(vm_id, vm_dir, config)
36

37
        self.instances[vm_id] = {
38
            'directory': vm_dir,
39
            'cgroup': cgroup_config,
40
            'command': jailer_cmd
41
        }
42

43
        return jailer_cmd
44

45
    def _setup_chroot_binaries(self, chroot_dir):
46
        """Set up minimal binaries in chroot environment"""
47

48
        # Copy Firecracker binary
49
        firecracker_src = '/usr/local/bin/firecracker'
50
        firecracker_dst = chroot_dir / 'bin' / 'firecracker'
51

52
        os.makedirs(firecracker_dst.parent, exist_ok=True)
53
        subprocess.run(['cp', firecracker_src, firecracker_dst], check=True)
54

55
        # Copy required libraries using ldd
56
        ldd_output = subprocess.run(
57
            ['ldd', firecracker_src],
58
            capture_output=True, text=True
59
        ).stdout
60

61
        for line in ldd_output.split('\n'):
62
            if '=>' in line and '/' in line:
63
                lib_path = line.split('=>')[1].strip().split()[0]
64
                if lib_path.startswith('/'):
65
                    lib_dst = chroot_dir / lib_path.lstrip('/')
66
                    lib_dst.parent.mkdir(parents=True, exist_ok=True)
67
                    subprocess.run(['cp', lib_path, lib_dst], check=True)
68

69
    def _create_device_nodes(self, dev_dir):
70
        """Create necessary device nodes in chroot"""
71

72
        devices = [
73
            ('null', 'c', 1, 3, 0o666),
74
            ('zero', 'c', 1, 5, 0o666),
75
            ('random', 'c', 1, 8, 0o666),
76
            ('urandom', 'c', 1, 9, 0o666),
77
            ('kvm', 'c', 10, 232, 0o660)
78
        ]
79

80
        for name, type_char, major, minor, mode in devices:
81
            device_path = dev_dir / name
82
            subprocess.run([
83
                'sudo', 'mknod', str(device_path),
84
                type_char, str(major), str(minor)
85
            ], check=True)
86
            subprocess.run(['sudo', 'chmod', oct(mode), str(device_path)], check=True)
87

88
    def _setup_cgroups(self, vm_id, resources):
89
        """Set up cgroups v2 for resource isolation"""
90

91
        cgroup_path = f"/sys/fs/cgroup/firecracker/{vm_id}"
92
        os.makedirs(cgroup_path, exist_ok=True)
93

94
        # Memory limit
95
        if 'memory_mb' in resources:
96
            with open(f"{cgroup_path}/memory.max", 'w') as f:
97
                f.write(str(resources['memory_mb'] * 1024 * 1024))
98

99
        # CPU limit
100
        if 'cpu_quota' in resources:
101
            with open(f"{cgroup_path}/cpu.max", 'w') as f:
102
                f.write(f"{resources['cpu_quota']} 100000")
103

104
        # I/O limits
105
        if 'io_weight' in resources:
106
            with open(f"{cgroup_path}/io.weight", 'w') as f:
107
                f.write(str(resources['io_weight']))
108

109
        return cgroup_path
110

111
    def _build_jailer_command(self, vm_id, vm_dir, config):
112
        """Build the jailer command with security options"""
113

114
        cmd = [
115
            'sudo', 'jailer',
116
            '--id', vm_id,
117
            '--exec-file', '/usr/local/bin/firecracker',
118
            '--uid', str(config.get('uid', 1000)),
119
            '--gid', str(config.get('gid', 1000)),
120
            '--chroot-base-dir', str(vm_dir),
121
            '--daemonize'
122
        ]
123

124
        # Network namespace isolation
125
        if config.get('new_netns', True):
126
            cmd.extend(['--new-pid-ns'])
127

128
        # Additional namespaces
129
        if config.get('new_pid_ns', True):
130
            cmd.extend(['--new-pid-ns'])
131

132
        # Resource constraints
133
        if 'numa_node' in config:
134
            cmd.extend(['--numa-node', str(config['numa_node'])])
135

136
        # Firecracker arguments
137
        cmd.append('--')
138
        cmd.extend(['--api-sock', 'socket'])
139

140
        if config.get('config_file'):
141
            cmd.extend(['--config-file', config['config_file']])
142

143
        return cmd
144

145
    def cleanup_instance(self, vm_id):
146
        """Clean up instance resources"""
147

148
        if vm_id in self.instances:
149
            instance = self.instances[vm_id]
150

151
            # Remove cgroup
152
            cgroup_path = instance['cgroup']
153
            if os.path.exists(cgroup_path):
154
                subprocess.run(['sudo', 'rmdir', cgroup_path], check=False)
155

156
            # Remove instance directory
157
            subprocess.run(['sudo', 'rm', '-rf', str(instance['directory'])], check=False)
158

159
            del self.instances[vm_id]
160

161
# Usage example
162
if __name__ == '__main__':
163
    jailer = SecureJailer()
164

165
    config = {
166
        'uid': 1000,
167
        'gid': 1000,
168
        'new_netns': True,
169
        'new_pid_ns': True,
170
        'numa_node': 0,
171
        'resources': {
172
            'memory_mb': 512,
173
            'cpu_quota': 50000,  # 50% of one CPU
174
            'io_weight': 100
175
        }
176
    }
177

178
    cmd = jailer.create_secure_environment('test-vm', config)
179
    print("Jailer command:")
180
    print(' '.join(cmd))

Seccomp-BPF Filtering#

Firecracker uses seccomp-BPF (Berkeley Packet Filter) to restrict system calls, dramatically reducing the attack surface.

Default Seccomp Profile#

1
{
2
  "seccomp": {
3
    "default_action": "kill_process",
4
    "filters": [
5
      {
6
        "syscalls": [
7
          "accept4",
8
          "brk",
9
          "clock_gettime",
10
          "close",
11
          "connect",
12
          "dup",
13
          "epoll_ctl",
14
          "epoll_pwait",
15
          "exit",
16
          "exit_group",
17
          "fcntl",
18
          "fstat",
19
          "futex",
20
          "getpid",
21
          "ioctl",
22
          "lseek",
23
          "mmap",
24
          "munmap",
25
          "open",
26
          "openat",
27
          "pipe",
28
          "prctl",
29
          "pread64",
30
          "pwrite64",
31
          "read",
32
          "readv",
33
          "recvfrom",
34
          "rt_sigaction",
35
          "rt_sigprocmask",
36
          "rt_sigreturn",
37
          "sendto",
38
          "sigaltstack",
39
          "socket",
40
          "stat",
41
          "timerfd_create",
42
          "timerfd_settime",
43
          "write",
44
          "writev"
45
        ],
46
        "action": "allow"
47
      },
48
      {
49
        "syscalls": ["tkill"],
50
        "action": "allow",
51
        "args": [
52
          {
53
            "index": 1,
54
            "type": "dword",
55
            "op": "eq",
56
            "val": 15,
57
            "comment": "SIGTERM"
58
          }
59
        ]
60
      }
61
    ]
62
  }
63
}

Custom Seccomp Profiles#

1
#!/usr/bin/env python3
2
import json
3

4
class SeccompProfileGenerator:
5
    """Generate custom seccomp profiles for Firecracker"""
6

7
    # Base required syscalls for Firecracker
8
    BASE_SYSCALLS = [
9
        "accept4", "brk", "clock_gettime", "close", "connect",
10
        "dup", "epoll_ctl", "epoll_pwait", "exit", "exit_group",
11
        "fcntl", "fstat", "futex", "getpid", "ioctl", "lseek",
12
        "mmap", "munmap", "open", "openat", "pipe", "prctl",
13
        "pread64", "pwrite64", "read", "readv", "recvfrom",
14
        "rt_sigaction", "rt_sigprocmask", "rt_sigreturn",
15
        "sendto", "sigaltstack", "socket", "stat",
16
        "timerfd_create", "timerfd_settime", "write", "writev"
17
    ]
18

19
    # Additional syscalls for specific features
20
    FEATURE_SYSCALLS = {
21
        'networking': ['bind', 'listen', 'getsockopt', 'setsockopt'],
22
        'file_operations': ['ftruncate', 'fsync', 'unlink', 'rename'],
23
        'memory_advanced': ['madvise', 'mprotect', 'mremap'],
24
        'process_control': ['clone', 'fork', 'execve', 'wait4'],
25
        'time_operations': ['gettimeofday', 'nanosleep', 'clock_nanosleep']
26
    }
27

28
    # Dangerous syscalls to never allow
29
    BLOCKED_SYSCALLS = [
30
        'ptrace', 'kexec_load', 'create_module', 'init_module',
31
        'delete_module', 'ioperm', 'iopl', 'mount', 'umount2',
32
        'swapon', 'swapoff', 'reboot', 'sethostname', 'setdomainname',
33
        'acct', 'chroot', 'pivot_root'
34
    ]
35

36
    def generate_profile(self, features=None, custom_syscalls=None):
37
        """Generate a seccomp profile with specified features"""
38

39
        if features is None:
40
            features = []
41
        if custom_syscalls is None:
42
            custom_syscalls = []
43

44
        # Start with base syscalls
45
        allowed_syscalls = self.BASE_SYSCALLS.copy()
46

47
        # Add feature-specific syscalls
48
        for feature in features:
49
            if feature in self.FEATURE_SYSCALLS:
50
                allowed_syscalls.extend(self.FEATURE_SYSCALLS[feature])
51

52
        # Add custom syscalls
53
        allowed_syscalls.extend(custom_syscalls)
54

55
        # Remove duplicates and blocked syscalls
56
        allowed_syscalls = list(set(allowed_syscalls))
57
        allowed_syscalls = [sc for sc in allowed_syscalls if sc not in self.BLOCKED_SYSCALLS]
58
        allowed_syscalls.sort()
59

60
        profile = {
61
            "seccomp": {
62
                "default_action": "kill_process",
63
                "filters": [
64
                    {
65
                        "syscalls": allowed_syscalls,
66
                        "action": "allow"
67
                    }
68
                ]
69
            }
70
        }
71

72
        return profile
73

74
    def generate_conditional_profile(self, conditions):
75
        """Generate profile with conditional syscall permissions"""
76

77
        filters = []
78

79
        # Base allow filter
80
        filters.append({
81
            "syscalls": self.BASE_SYSCALLS,
82
            "action": "allow"
83
        })
84

85
        # Add conditional filters
86
        for condition in conditions:
87
            syscall = condition['syscall']
88
            args = condition.get('args', [])
89
            action = condition.get('action', 'allow')
90

91
            filter_entry = {
92
                "syscalls": [syscall],
93
                "action": action
94
            }
95

96
            if args:
97
                filter_entry['args'] = args
98

99
            filters.append(filter_entry)
100

101
        profile = {
102
            "seccomp": {
103
                "default_action": "kill_process",
104
                "filters": filters
105
            }
106
        }
107

108
        return profile
109

110
    def validate_profile(self, profile):
111
        """Validate seccomp profile structure"""
112

113
        required_keys = ['seccomp']
114
        seccomp_keys = ['default_action', 'filters']
115

116
        if not all(key in profile for key in required_keys):
117
            return False, "Missing required top-level keys"
118

119
        seccomp = profile['seccomp']
120
        if not all(key in seccomp for key in seccomp_keys):
121
            return False, "Missing required seccomp keys"
122

123
        # Check for dangerous syscalls
124
        all_syscalls = []
125
        for filter_entry in seccomp['filters']:
126
            if 'syscalls' in filter_entry:
127
                all_syscalls.extend(filter_entry['syscalls'])
128

129
        dangerous_found = [sc for sc in all_syscalls if sc in self.BLOCKED_SYSCALLS]
130
        if dangerous_found:
131
            return False, f"Dangerous syscalls found: {dangerous_found}"
132

133
        return True, "Profile is valid"
134

135
# Usage examples
136
if __name__ == '__main__':
137
    generator = SeccompProfileGenerator()
138

139
    # Generate basic profile
140
    basic_profile = generator.generate_profile()
141
    print("Basic Profile:")
142
    print(json.dumps(basic_profile, indent=2))
143

144
    # Generate profile with networking features
145
    network_profile = generator.generate_profile(features=['networking'])
146
    print("\nNetworking Profile:")
147
    print(json.dumps(network_profile, indent=2))
148

149
    # Generate conditional profile
150
    conditions = [
151
        {
152
            'syscall': 'tkill',
153
            'action': 'allow',
154
            'args': [
155
                {
156
                    'index': 1,
157
                    'type': 'dword',
158
                    'op': 'eq',
159
                    'val': 15,
160
                    'comment': 'SIGTERM'
161
                }
162
            ]
163
        }
164
    ]
165

166
    conditional_profile = generator.generate_conditional_profile(conditions)
167
    print("\nConditional Profile:")
168
    print(json.dumps(conditional_profile, indent=2))
169

170
    # Validate profile
171
    valid, message = generator.validate_profile(basic_profile)
172
    print(f"\nProfile validation: {valid} - {message}")

Applying Custom Seccomp Profiles#

1
#!/bin/bash
2

3
# Create custom seccomp profile
4
cat > custom_seccomp.json << 'EOF'
5
{
6
  "seccomp": {
7
    "default_action": "kill_process",
8
    "filters": [
9
      {
10
        "syscalls": [
11
          "accept4", "brk", "clock_gettime", "close", "connect",
12
          "dup", "epoll_ctl", "epoll_pwait", "exit", "exit_group",
13
          "fcntl", "fstat", "futex", "getpid", "ioctl", "lseek",
14
          "mmap", "munmap", "open", "openat", "pipe", "prctl",
15
          "pread64", "pwrite64", "read", "readv", "recvfrom",
16
          "rt_sigaction", "rt_sigprocmask", "rt_sigreturn",
17
          "sendto", "sigaltstack", "socket", "stat",
18
          "timerfd_create", "timerfd_settime", "write", "writev"
19
        ],
20
        "action": "allow"
21
      },
22
      {
23
        "syscalls": ["bind", "listen"],
24
        "action": "allow",
25
        "comment": "Allow networking syscalls"
26
      }
27
    ]
28
  }
29
}
30
EOF
31

32
# Run Firecracker with custom seccomp profile
33
firecracker \
34
  --api-sock /tmp/firecracker.socket \
35
  --seccomp-filter custom_seccomp.json \
36
  --config-file vm_config.json

Resource Isolation with Cgroups#

Cgroups provide fine-grained resource control and isolation for Firecracker processes.

Cgroups v2 Configuration#

1
#!/usr/bin/env python3
2
import os
3
import subprocess
4
from pathlib import Path
5

6
class CgroupManager:
7
    """Manage cgroups v2 for Firecracker instances"""
8

9
    def __init__(self, base_path="/sys/fs/cgroup/firecracker"):
10
        self.base_path = Path(base_path)
11
        self.base_path.mkdir(parents=True, exist_ok=True)
12

13
        # Enable required controllers
14
        self._enable_controllers(['cpu', 'memory', 'io', 'pids'])
15

16
    def _enable_controllers(self, controllers):
17
        """Enable cgroup controllers"""
18

19
        try:
20
            # Enable controllers in parent cgroup
21
            parent_controllers = self.base_path.parent / "cgroup.subtree_control"
22
            if parent_controllers.exists():
23
                with open(parent_controllers, 'w') as f:
24
                    f.write(' '.join(f'+{c}' for c in controllers))
25
        except PermissionError:
26
            print(f"Warning: Cannot enable controllers. Run with sudo or check permissions.")
27

28
    def create_vm_cgroup(self, vm_id, limits):
29
        """Create cgroup for a specific VM with resource limits"""
30

31
        vm_cgroup = self.base_path / vm_id
32
        vm_cgroup.mkdir(exist_ok=True)
33

34
        # Set memory limits
35
        if 'memory_mb' in limits:
36
            self._set_memory_limit(vm_cgroup, limits['memory_mb'])
37

38
        # Set CPU limits
39
        if 'cpu_quota' in limits:
40
            self._set_cpu_limit(vm_cgroup, limits['cpu_quota'])
41

42
        # Set I/O limits
43
        if 'io_limits' in limits:
44
            self._set_io_limits(vm_cgroup, limits['io_limits'])
45

46
        # Set process limits
47
        if 'max_pids' in limits:
48
            self._set_pid_limit(vm_cgroup, limits['max_pids'])
49

50
        return str(vm_cgroup)
51

52
    def _set_memory_limit(self, cgroup_path, memory_mb):
53
        """Set memory limits for cgroup"""
54

55
        memory_bytes = memory_mb * 1024 * 1024
56

57
        # Hard memory limit
58
        memory_max = cgroup_path / "memory.max"
59
        with open(memory_max, 'w') as f:
60
            f.write(str(memory_bytes))
61

62
        # Soft memory limit (for memory pressure)
63
        memory_high = cgroup_path / "memory.high"
64
        with open(memory_high, 'w') as f:
65
            f.write(str(int(memory_bytes * 0.9)))  # 90% of max
66

67
        # Swap limit
68
        memory_swap_max = cgroup_path / "memory.swap.max"
69
        if memory_swap_max.exists():
70
            with open(memory_swap_max, 'w') as f:
71
                f.write(str(memory_bytes // 2))  # 50% of memory as swap
72

73
    def _set_cpu_limit(self, cgroup_path, cpu_quota_us):
74
        """Set CPU limits for cgroup"""
75

76
        # CPU quota in microseconds per 100ms period
77
        cpu_max = cgroup_path / "cpu.max"
78
        with open(cpu_max, 'w') as f:
79
            f.write(f"{cpu_quota_us} 100000")
80

81
        # CPU weight (relative priority)
82
        cpu_weight = cgroup_path / "cpu.weight"
83
        with open(cpu_weight, 'w') as f:
84
            f.write("100")  # Default weight
85

86
    def _set_io_limits(self, cgroup_path, io_limits):
87
        """Set I/O limits for cgroup"""
88

89
        # I/O weight
90
        if 'weight' in io_limits:
91
            io_weight = cgroup_path / "io.weight"
92
            with open(io_weight, 'w') as f:
93
                f.write(str(io_limits['weight']))
94

95
        # I/O bandwidth limits
96
        if 'bps_limits' in io_limits:
97
            io_max = cgroup_path / "io.max"
98
            with open(io_max, 'w') as f:
99
                for device, limits_dict in io_limits['bps_limits'].items():
100
                    rbps = limits_dict.get('rbps', 'max')
101
                    wbps = limits_dict.get('wbps', 'max')
102
                    riops = limits_dict.get('riops', 'max')
103
                    wiops = limits_dict.get('wiops', 'max')
104
                    f.write(f"{device} rbps={rbps} wbps={wbps} riops={riops} wiops={wiops}\n")
105

106
    def _set_pid_limit(self, cgroup_path, max_pids):
107
        """Set process limit for cgroup"""
108

109
        pids_max = cgroup_path / "pids.max"
110
        with open(pids_max, 'w') as f:
111
            f.write(str(max_pids))
112

113
    def add_process_to_cgroup(self, cgroup_path, pid):
114
        """Add process to cgroup"""
115

116
        procs_file = Path(cgroup_path) / "cgroup.procs"
117
        with open(procs_file, 'w') as f:
118
            f.write(str(pid))
119

120
    def get_cgroup_stats(self, cgroup_path):
121
        """Get cgroup resource usage statistics"""
122

123
        stats = {}
124
        cgroup_path = Path(cgroup_path)
125

126
        # Memory stats
127
        memory_current = cgroup_path / "memory.current"
128
        if memory_current.exists():
129
            with open(memory_current, 'r') as f:
130
                stats['memory_current'] = int(f.read().strip())
131

132
        memory_stat = cgroup_path / "memory.stat"
133
        if memory_stat.exists():
134
            memory_details = {}
135
            with open(memory_stat, 'r') as f:
136
                for line in f:
137
                    key, value = line.strip().split()
138
                    memory_details[key] = int(value)
139
            stats['memory_details'] = memory_details
140

141
        # CPU stats
142
        cpu_stat = cgroup_path / "cpu.stat"
143
        if cpu_stat.exists():
144
            cpu_details = {}
145
            with open(cpu_stat, 'r') as f:
146
                for line in f:
147
                    key, value = line.strip().split()
148
                    cpu_details[key] = int(value)
149
            stats['cpu_details'] = cpu_details
150

151
        # I/O stats
152
        io_stat = cgroup_path / "io.stat"
153
        if io_stat.exists():
154
            io_details = {}
155
            with open(io_stat, 'r') as f:
156
                for line in f:
157
                    parts = line.strip().split()
158
                    device = parts[0]
159
                    device_stats = {}
160
                    for stat in parts[1:]:
161
                        key, value = stat.split('=')
162
                        device_stats[key] = int(value)
163
                    io_details[device] = device_stats
164
            stats['io_details'] = io_details
165

166
        # PID stats
167
        pids_current = cgroup_path / "pids.current"
168
        if pids_current.exists():
169
            with open(pids_current, 'r') as f:
170
                stats['pids_current'] = int(f.read().strip())
171

172
        return stats
173

174
    def cleanup_cgroup(self, vm_id):
175
        """Remove VM cgroup"""
176

177
        vm_cgroup = self.base_path / vm_id
178
        if vm_cgroup.exists():
179
            try:
180
                # Kill all processes in cgroup first
181
                procs_file = vm_cgroup / "cgroup.procs"
182
                if procs_file.exists():
183
                    with open(procs_file, 'r') as f:
184
                        pids = f.read().strip().split('\n')
185

186
                    for pid in pids:
187
                        if pid:
188
                            try:
189
                                os.kill(int(pid), 15)  # SIGTERM
190
                            except (ProcessLookupError, ValueError):
191
                                pass
192

193
                # Remove cgroup directory
194
                vm_cgroup.rmdir()
195

196
            except Exception as e:
197
                print(f"Error cleaning up cgroup {vm_id}: {e}")
198

199
# Usage example
200
if __name__ == '__main__':
201
    manager = CgroupManager()
202

203
    # Create cgroup with limits
204
    limits = {
205
        'memory_mb': 512,
206
        'cpu_quota': 50000,  # 50% of one CPU
207
        'io_limits': {
208
            'weight': 100,
209
            'bps_limits': {
210
                '8:0': {  # First SCSI disk
211
                    'rbps': 10485760,  # 10MB/s read
212
                    'wbps': 5242880,   # 5MB/s write
213
                    'riops': 1000,     # 1000 read IOPS
214
                    'wiops': 500       # 500 write IOPS
215
                }
216
            }
217
        },
218
        'max_pids': 100
219
    }
220

221
    cgroup_path = manager.create_vm_cgroup('test-vm', limits)
222
    print(f"Created cgroup: {cgroup_path}")
223

224
    # Simulate adding a process
225
    # manager.add_process_to_cgroup(cgroup_path, os.getpid())
226

227
    # Get stats
228
    stats = manager.get_cgroup_stats(cgroup_path)
229
    print(f"Cgroup stats: {stats}")
230

231
    # Cleanup
232
    # manager.cleanup_cgroup('test-vm')

Network Security#

Network Namespace Isolation#

1
#!/bin/bash
2

3
# Create network namespace for VM isolation
4
create_vm_network() {
5
    local vm_id=$1
6
    local vm_ip=$2
7

8
    # Create network namespace
9
    sudo ip netns add "fc-${vm_id}"
10

11
    # Create veth pair
12
    sudo ip link add "veth-${vm_id}" type veth peer name "veth-${vm_id}-peer"
13

14
    # Move peer to namespace
15
    sudo ip link set "veth-${vm_id}-peer" netns "fc-${vm_id}"
16

17
    # Configure host side
18
    sudo ip addr add "192.168.100.1/24" dev "veth-${vm_id}"
19
    sudo ip link set "veth-${vm_id}" up
20

21
    # Configure namespace side
22
    sudo ip netns exec "fc-${vm_id}" ip addr add "${vm_ip}/24" dev "veth-${vm_id}-peer"
23
    sudo ip netns exec "fc-${vm_id}" ip link set "veth-${vm_id}-peer" up
24
    sudo ip netns exec "fc-${vm_id}" ip link set lo up
25
    sudo ip netns exec "fc-${vm_id}" ip route add default via 192.168.100.1
26

27
    # Create TAP interface in namespace
28
    sudo ip netns exec "fc-${vm_id}" ip tuntap add "tap-${vm_id}" mode tap
29
    sudo ip netns exec "fc-${vm_id}" ip link set "tap-${vm_id}" up
30

31
    # Bridge TAP to veth
32
    sudo ip netns exec "fc-${vm_id}" brctl addbr "br-${vm_id}"
33
    sudo ip netns exec "fc-${vm_id}" brctl addif "br-${vm_id}" "tap-${vm_id}"
34
    sudo ip netns exec "fc-${vm_id}" brctl addif "br-${vm_id}" "veth-${vm_id}-peer"
35
    sudo ip netns exec "fc-${vm_id}" ip link set "br-${vm_id}" up
36

37
    echo "Created network namespace fc-${vm_id} with IP ${vm_ip}"
38
}
39

40
# Usage
41
create_vm_network "vm001" "192.168.100.10"

Network Security Policies#

1
#!/usr/bin/env python3
2
import subprocess
3
import json
4
from dataclasses import dataclass
5
from typing import List, Dict, Optional
6

7
@dataclass
8
class NetworkRule:
9
    direction: str  # 'ingress' or 'egress'
10
    protocol: str   # 'tcp', 'udp', 'icmp'
11
    port_range: Optional[str] = None
12
    source_cidr: Optional[str] = None
13
    dest_cidr: Optional[str] = None
14
    action: str = 'ACCEPT'  # 'ACCEPT' or 'DROP'
15

16
class NetworkSecurityManager:
17
    """Manage network security for Firecracker VMs"""
18

19
    def __init__(self):
20
        self.vm_rules = {}
21

22
    def create_vm_network_policy(self, vm_id: str, rules: List[NetworkRule]):
23
        """Create iptables rules for VM network isolation"""
24

25
        self.vm_rules[vm_id] = rules
26

27
        # Create custom chains for this VM
28
        self._create_vm_chains(vm_id)
29

30
        # Apply rules
31
        for rule in rules:
32
            self._apply_rule(vm_id, rule)
33

34
        # Add jump rules to main chains
35
        self._add_jump_rules(vm_id)
36

37
    def _create_vm_chains(self, vm_id: str):
38
        """Create custom iptables chains for VM"""
39

40
        chains = [f'FC-{vm_id}-IN', f'FC-{vm_id}-OUT']
41

42
        for chain in chains:
43
            # Create chain
44
            subprocess.run(['sudo', 'iptables', '-t', 'filter', '-N', chain],
45
                          check=False)
46

47
            # Flush existing rules
48
            subprocess.run(['sudo', 'iptables', '-t', 'filter', '-F', chain])
49

50
    def _apply_rule(self, vm_id: str, rule: NetworkRule):
51
        """Apply a single network rule"""
52

53
        chain = f'FC-{vm_id}-IN' if rule.direction == 'ingress' else f'FC-{vm_id}-OUT'
54

55
        cmd = ['sudo', 'iptables', '-t', 'filter', '-A', chain]
56

57
        # Protocol
58
        cmd.extend(['-p', rule.protocol])
59

60
        # Port range
61
        if rule.port_range:
62
            if rule.direction == 'ingress':
63
                cmd.extend(['--dport', rule.port_range])
64
            else:
65
                cmd.extend(['--sport', rule.port_range])
66

67
        # Source/destination CIDR
68
        if rule.source_cidr:
69
            cmd.extend(['-s', rule.source_cidr])
70

71
        if rule.dest_cidr:
72
            cmd.extend(['-d', rule.dest_cidr])
73

74
        # Action
75
        cmd.extend(['-j', rule.action])
76

77
        subprocess.run(cmd, check=True)
78

79
    def _add_jump_rules(self, vm_id: str):
80
        """Add jump rules to custom chains"""
81

82
        vm_interface = f'tap-{vm_id}'
83

84
        # Ingress traffic
85
        subprocess.run([
86
            'sudo', 'iptables', '-t', 'filter', '-I', 'FORWARD',
87
            '-i', vm_interface, '-j', f'FC-{vm_id}-IN'
88
        ], check=False)
89

90
        # Egress traffic
91
        subprocess.run([
92
            'sudo', 'iptables', '-t', 'filter', '-I', 'FORWARD',
93
            '-o', vm_interface, '-j', f'FC-{vm_id}-OUT'
94
        ], check=False)
95

96
    def create_default_security_policy(self, vm_id: str,
97
                                     allowed_ports: List[int] = None,
98
                                     allowed_cidrs: List[str] = None):
99
        """Create a default security policy for a VM"""
100

101
        if allowed_ports is None:
102
            allowed_ports = [22, 80, 443]  # SSH, HTTP, HTTPS
103

104
        if allowed_cidrs is None:
105
            allowed_cidrs = ['10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16']
106

107
        rules = []
108

109
        # Allow established connections
110
        rules.append(NetworkRule(
111
            direction='ingress',
112
            protocol='tcp',
113
            action='ACCEPT'
114
        ))
115

116
        # Allow specific ports from allowed CIDRs
117
        for port in allowed_ports:
118
            for cidr in allowed_cidrs:
119
                rules.append(NetworkRule(
120
                    direction='ingress',
121
                    protocol='tcp',
122
                    port_range=str(port),
123
                    source_cidr=cidr,
124
                    action='ACCEPT'
125
                ))
126

127
        # Allow outgoing connections to allowed CIDRs
128
        for cidr in allowed_cidrs:
129
            rules.append(NetworkRule(
130
                direction='egress',
131
                protocol='tcp',
132
                dest_cidr=cidr,
133
                action='ACCEPT'
134
            ))
135

136
        # Allow DNS
137
        rules.extend([
138
            NetworkRule(
139
                direction='egress',
140
                protocol='udp',
141
                port_range='53',
142
                action='ACCEPT'
143
            ),
144
            NetworkRule(
145
                direction='egress',
146
                protocol='tcp',
147
                port_range='53',
148
                action='ACCEPT'
149
            )
150
        ])
151

152
        # Allow ICMP
153
        rules.append(NetworkRule(
154
            direction='ingress',
155
            protocol='icmp',
156
            action='ACCEPT'
157
        ))
158

159
        # Default deny
160
        rules.extend([
161
            NetworkRule(
162
                direction='ingress',
163
                protocol='all',
164
                action='DROP'
165
            ),
166
            NetworkRule(
167
                direction='egress',
168
                protocol='all',
169
                action='DROP'
170
            )
171
        ])
172

173
        self.create_vm_network_policy(vm_id, rules)
174

175
    def cleanup_vm_policy(self, vm_id: str):
176
        """Remove network policy for VM"""
177

178
        chains = [f'FC-{vm_id}-IN', f'FC-{vm_id}-OUT']
179

180
        # Remove jump rules
181
        vm_interface = f'tap-{vm_id}'
182
        subprocess.run([
183
            'sudo', 'iptables', '-t', 'filter', '-D', 'FORWARD',
184
            '-i', vm_interface, '-j', f'FC-{vm_id}-IN'
185
        ], check=False)
186

187
        subprocess.run([
188
            'sudo', 'iptables', '-t', 'filter', '-D', 'FORWARD',
189
            '-o', vm_interface, '-j', f'FC-{vm_id}-OUT'
190
        ], check=False)
191

192
        # Flush and delete chains
193
        for chain in chains:
194
            subprocess.run(['sudo', 'iptables', '-t', 'filter', '-F', chain],
195
                          check=False)
196
            subprocess.run(['sudo', 'iptables', '-t', 'filter', '-X', chain],
197
                          check=False)
198

199
        if vm_id in self.vm_rules:
200
            del self.vm_rules[vm_id]
201

202
    def get_policy_status(self, vm_id: str):
203
        """Get current policy status for VM"""
204

205
        result = subprocess.run([
206
            'sudo', 'iptables', '-t', 'filter', '-L', f'FC-{vm_id}-IN', '-n'
207
        ], capture_output=True, text=True, check=False)
208

209
        if result.returncode == 0:
210
            return {
211
                'status': 'active',
212
                'rules': result.stdout,
213
                'vm_id': vm_id
214
            }
215
        else:
216
            return {
217
                'status': 'not_found',
218
                'vm_id': vm_id
219
            }
220

221
# Usage example
222
if __name__ == '__main__':
223
    manager = NetworkSecurityManager()
224

225
    # Create default security policy
226
    manager.create_default_security_policy('vm001')
227

228
    # Check status
229
    status = manager.get_policy_status('vm001')
230
    print(f"Policy status: {status}")
231

232
    # Custom rules example
233
    custom_rules = [
234
        NetworkRule(
235
            direction='ingress',
236
            protocol='tcp',
237
            port_range='8080',
238
            source_cidr='192.168.1.0/24',
239
            action='ACCEPT'
240
        ),
241
        NetworkRule(
242
            direction='egress',
243
            protocol='tcp',
244
            port_range='443',
245
            dest_cidr='0.0.0.0/0',
246
            action='ACCEPT'
247
        )
248
    ]
249

250
    manager.create_vm_network_policy('vm002', custom_rules)
251

252
    # Cleanup
253
    # manager.cleanup_vm_policy('vm001')
254
    # manager.cleanup_vm_policy('vm002')

Security Monitoring and Auditing#

Security Event Monitoring#

1
#!/usr/bin/env python3
2
import json
3
import time
4
import subprocess
5
import logging
6
from datetime import datetime
7
from pathlib import Path
8

9
class SecurityMonitor:
10
    """Monitor security events for Firecracker VMs"""
11

12
    def __init__(self, log_file='/var/log/firecracker-security.log'):
13
        self.log_file = log_file
14
        self.setup_logging()
15
        self.monitored_events = [
16
            'syscall_violations',
17
            'memory_violations',
18
            'network_violations',
19
            'resource_violations',
20
            'process_violations'
21
        ]
22

23
    def setup_logging(self):
24
        """Set up security logging"""
25

26
        logging.basicConfig(
27
            level=logging.INFO,
28
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
29
            handlers=[
30
                logging.FileHandler(self.log_file),
31
                logging.StreamHandler()
32
            ]
33
        )
34

35
        self.logger = logging.getLogger('firecracker-security')
36

37
    def monitor_seccomp_violations(self, vm_id):
38
        """Monitor for seccomp violations"""
39

40
        try:
41
            # Monitor kernel messages for seccomp violations
42
            result = subprocess.run([
43
                'dmesg', '-T', '--follow'
44
            ], capture_output=True, text=True, timeout=1)
45

46
            for line in result.stdout.split('\n'):
47
                if 'seccomp' in line.lower() and 'SIGKILL' in line:
48
                    self.log_security_event(vm_id, 'seccomp_violation', {
49
                        'message': line,
50
                        'timestamp': datetime.now().isoformat(),
51
                        'severity': 'critical'
52
                    })
53

54
        except subprocess.TimeoutExpired:
55
            pass
56
        except Exception as e:
57
            self.logger.error(f"Error monitoring seccomp: {e}")
58

59
    def monitor_memory_violations(self, vm_id, cgroup_path):
60
        """Monitor for memory violations"""
61

62
        try:
63
            memory_events = Path(cgroup_path) / "memory.events"
64
            if memory_events.exists():
65
                with open(memory_events, 'r') as f:
66
                    events = {}
67
                    for line in f:
68
                        key, value = line.strip().split()
69
                        events[key] = int(value)
70

71
                # Check for OOM events
72
                if events.get('oom', 0) > 0:
73
                    self.log_security_event(vm_id, 'memory_oom', {
74
                        'oom_count': events['oom'],
75
                        'oom_kill': events.get('oom_kill', 0),
76
                        'timestamp': datetime.now().isoformat(),
77
                        'severity': 'high'
78
                    })
79

80
                # Check for memory pressure
81
                if events.get('high', 0) > 0:
82
                    self.log_security_event(vm_id, 'memory_pressure', {
83
                        'high_events': events['high'],
84
                        'max_events': events.get('max', 0),
85
                        'timestamp': datetime.now().isoformat(),
86
                        'severity': 'medium'
87
                    })
88

89
        except Exception as e:
90
            self.logger.error(f"Error monitoring memory: {e}")
91

92
    def monitor_network_violations(self, vm_id):
93
        """Monitor for network security violations"""
94

95
        try:
96
            # Check iptables logs for dropped packets
97
            result = subprocess.run([
98
                'journalctl', '-u', 'iptables', '-n', '10', '--no-pager'
99
            ], capture_output=True, text=True, check=False)
100

101
            for line in result.stdout.split('\n'):
102
                if f'FC-{vm_id}' in line and 'DROP' in line:
103
                    self.log_security_event(vm_id, 'network_violation', {
104
                        'message': line,
105
                        'timestamp': datetime.now().isoformat(),
106
                        'severity': 'medium'
107
                    })
108

109
        except Exception as e:
110
            self.logger.error(f"Error monitoring network: {e}")
111

112
    def monitor_resource_violations(self, vm_id, cgroup_path):
113
        """Monitor for resource limit violations"""
114

115
        try:
116
            cgroup_path = Path(cgroup_path)
117

118
            # Check CPU throttling
119
            cpu_stat = cgroup_path / "cpu.stat"
120
            if cpu_stat.exists():
121
                with open(cpu_stat, 'r') as f:
122
                    cpu_stats = {}
123
                    for line in f:
124
                        key, value = line.strip().split()
125
                        cpu_stats[key] = int(value)
126

127
                if cpu_stats.get('throttled_time', 0) > 0:
128
                    self.log_security_event(vm_id, 'cpu_throttling', {
129
                        'throttled_time': cpu_stats['throttled_time'],
130
                        'throttled_periods': cpu_stats.get('throttled_periods', 0),
131
                        'timestamp': datetime.now().isoformat(),
132
                        'severity': 'low'
133
                    })
134

135
            # Check PID limits
136
            pids_events = cgroup_path / "pids.events"
137
            if pids_events.exists():
138
                with open(pids_events, 'r') as f:
139
                    for line in f:
140
                        key, value = line.strip().split()
141
                        if key == 'max' and int(value) > 0:
142
                            self.log_security_event(vm_id, 'pid_limit_hit', {
143
                                'max_events': int(value),
144
                                'timestamp': datetime.now().isoformat(),
145
                                'severity': 'medium'
146
                            })
147

148
        except Exception as e:
149
            self.logger.error(f"Error monitoring resources: {e}")
150

151
    def log_security_event(self, vm_id, event_type, details):
152
        """Log a security event"""
153

154
        event = {
155
            'vm_id': vm_id,
156
            'event_type': event_type,
157
            'details': details,
158
            'logged_at': datetime.now().isoformat()
159
        }
160

161
        self.logger.warning(f"Security event: {json.dumps(event)}")
162

163
        # Store in structured format for analysis
164
        self.store_event(event)
165

166
    def store_event(self, event):
167
        """Store security event for analysis"""
168

169
        events_file = Path(self.log_file).parent / "security_events.jsonl"
170

171
        with open(events_file, 'a') as f:
172
            f.write(json.dumps(event) + '\n')
173

174
    def get_security_summary(self, vm_id, hours=24):
175
        """Get security event summary for VM"""
176

177
        events_file = Path(self.log_file).parent / "security_events.jsonl"
178

179
        if not events_file.exists():
180
            return {'vm_id': vm_id, 'events': [], 'summary': {}}
181

182
        cutoff_time = datetime.now().timestamp() - (hours * 3600)
183
        events = []
184
        summary = {}
185

186
        with open(events_file, 'r') as f:
187
            for line in f:
188
                try:
189
                    event = json.loads(line.strip())
190

191
                    if event['vm_id'] != vm_id:
192
                        continue
193

194
                    event_time = datetime.fromisoformat(event['logged_at']).timestamp()
195
                    if event_time < cutoff_time:
196
                        continue
197

198
                    events.append(event)
199

200
                    event_type = event['event_type']
201
                    if event_type not in summary:
202
                        summary[event_type] = 0
203
                    summary[event_type] += 1
204

205
                except json.JSONDecodeError:
206
                    continue
207

208
        return {
209
            'vm_id': vm_id,
210
            'events': events,
211
            'summary': summary,
212
            'total_events': len(events)
213
        }
214

215
    def start_monitoring(self, vm_id, cgroup_path, interval=5):
216
        """Start continuous monitoring for a VM"""
217

218
        self.logger.info(f"Starting security monitoring for VM {vm_id}")
219

220
        while True:
221
            try:
222
                self.monitor_seccomp_violations(vm_id)
223
                self.monitor_memory_violations(vm_id, cgroup_path)
224
                self.monitor_network_violations(vm_id)
225
                self.monitor_resource_violations(vm_id, cgroup_path)
226

227
                time.sleep(interval)
228

229
            except KeyboardInterrupt:
230
                self.logger.info(f"Stopping monitoring for VM {vm_id}")
231
                break
232
            except Exception as e:
233
                self.logger.error(f"Error in monitoring loop: {e}")
234
                time.sleep(interval)
235

236
# Usage example
237
if __name__ == '__main__':
238
    monitor = SecurityMonitor()
239

240
    # Start monitoring for a VM
241
    vm_id = 'vm001'
242
    cgroup_path = f'/sys/fs/cgroup/firecracker/{vm_id}'
243

244
    try:
245
        monitor.start_monitoring(vm_id, cgroup_path, interval=5)
246
    except KeyboardInterrupt:
247
        print("Monitoring stopped")
248

249
    # Get security summary
250
    summary = monitor.get_security_summary(vm_id, hours=1)
251
    print(f"Security summary: {json.dumps(summary, indent=2)}")

Security Best Practices#

Production Security Checklist#

1
#!/bin/bash
2

3
# Firecracker Security Hardening Checklist
4

5
echo "=== Firecracker Security Hardening ==="
6

7
# 1. System-level security
8
echo "1. Checking system security..."
9

10
# Verify KVM permissions
11
if [ -c /dev/kvm ]; then
12
    echo "✓ /dev/kvm exists"
13
    ls -la /dev/kvm
14
else
15
    echo "✗ /dev/kvm not found"
16
fi
17

18
# Check kernel security features
19
echo "Checking kernel security features:"
20
grep -q smep /proc/cpuinfo && echo "✓ SMEP enabled" || echo "✗ SMEP not available"
21
grep -q smap /proc/cpuinfo && echo "✓ SMAP enabled" || echo "✗ SMAP not available"
22
grep -q pti /proc/cpuinfo && echo "✓ PTI available" || echo "✗ PTI not available"
23

24
# Check KASLR
25
grep -q nokaslr /proc/cmdline && echo "✗ KASLR disabled" || echo "✓ KASLR enabled"
26

27
# 2. Firecracker binary security
28
echo -e "\n2. Checking Firecracker binary..."
29

30
# Verify binary permissions
31
firecracker_bin=$(which firecracker)
32
if [ -f "$firecracker_bin" ]; then
33
    echo "✓ Firecracker binary found at $firecracker_bin"
34

35
    # Check file permissions
36
    perms=$(stat -c "%a" "$firecracker_bin")
37
    if [ "$perms" = "755" ]; then
38
        echo "✓ Correct permissions (755)"
39
    else
40
        echo "⚠ Permissions: $perms (should be 755)"
41
    fi
42

43
    # Check if PIE enabled
44
    if readelf -h "$firecracker_bin" | grep -q "DYN"; then
45
        echo "✓ PIE (Position Independent Executable) enabled"
46
    else
47
        echo "✗ PIE not enabled"
48
    fi
49

50
    # Check for stack canaries
51
    if readelf -s "$firecracker_bin" | grep -q "__stack_chk_fail"; then
52
        echo "✓ Stack canaries enabled"
53
    else
54
        echo "✗ Stack canaries not found"
55
    fi
56
else
57
    echo "✗ Firecracker binary not found"
58
fi
59

60
# 3. Directory security
61
echo -e "\n3. Checking directory security..."
62

63
# Check jailer workspace
64
jailer_dir="/srv/jailer"
65
if [ -d "$jailer_dir" ]; then
66
    echo "✓ Jailer directory exists: $jailer_dir"
67

68
    # Check ownership
69
    owner=$(stat -c "%U:%G" "$jailer_dir")
70
    echo "Directory owner: $owner"
71

72
    # Check permissions
73
    perms=$(stat -c "%a" "$jailer_dir")
74
    echo "Directory permissions: $perms"
75
else
76
    echo "✗ Jailer directory not found: $jailer_dir"
77
    echo "Creating jailer directory..."
78
    sudo mkdir -p "$jailer_dir"
79
    sudo chown root:root "$jailer_dir"
80
    sudo chmod 755 "$jailer_dir"
81
fi
82

83
# 4. Network security
84
echo -e "\n4. Checking network security..."
85

86
# Check IP forwarding
87
if sysctl net.ipv4.ip_forward | grep -q "= 1"; then
88
    echo "✓ IP forwarding enabled"
89
else
90
    echo "⚠ IP forwarding disabled (may need to enable)"
91
fi
92

93
# Check iptables
94
if iptables -L INPUT | grep -q "Chain INPUT"; then
95
    echo "✓ iptables is active"
96
else
97
    echo "✗ iptables not found"
98
fi
99

100
# 5. Resource limits
101
echo -e "\n5. Checking resource limits..."
102

103
# Check cgroups v2
104
if [ -d "/sys/fs/cgroup/cgroup.controllers" ]; then
105
    echo "✓ cgroups v2 available"
106
    available_controllers=$(cat /sys/fs/cgroup/cgroup.controllers)
107
    echo "Available controllers: $available_controllers"
108
else
109
    echo "✗ cgroups v2 not available"
110
fi
111

112
# Check ulimits
113
echo "Current ulimits:"
114
echo "  Max processes: $(ulimit -u)"
115
echo "  Max open files: $(ulimit -n)"
116
echo "  Max memory size: $(ulimit -m)"
117

118
# 6. Audit logging
119
echo -e "\n6. Checking audit configuration..."
120

121
# Check if auditd is running
122
if systemctl is-active --quiet auditd; then
123
    echo "✓ auditd is running"
124

125
    # Check if KVM syscalls are being audited
126
    if auditctl -l | grep -q kvm; then
127
        echo "✓ KVM syscalls are being audited"
128
    else
129
        echo "⚠ KVM syscalls not in audit rules"
130
        echo "Consider adding: auditctl -a always,exit -F arch=b64 -S ioctl -F path=/dev/kvm"
131
    fi
132
else
133
    echo "⚠ auditd is not running"
134
fi
135

136
echo -e "\n=== Security Hardening Complete ==="

Automated Security Validation#

1
#!/usr/bin/env python3
2
import os
3
import subprocess
4
import json
5
import time
6
from typing import Dict, List, Tuple
7

8
class SecurityValidator:
9
    """Validate Firecracker security configuration"""
10

11
    def __init__(self):
12
        self.validation_results = {}
13

14
    def validate_all(self, vm_id: str, config: Dict) -> Dict:
15
        """Run all security validations"""
16

17
        results = {
18
            'vm_id': vm_id,
19
            'timestamp': time.time(),
20
            'validations': {}
21
        }
22

23
        # Run individual validations
24
        results['validations']['jailer'] = self.validate_jailer_config(config.get('jailer', {}))
25
        results['validations']['seccomp'] = self.validate_seccomp_config(config.get('seccomp', {}))
26
        results['validations']['cgroups'] = self.validate_cgroup_config(config.get('cgroups', {}))
27
        results['validations']['network'] = self.validate_network_security(config.get('network', {}))
28
        results['validations']['resources'] = self.validate_resource_limits(config.get('resources', {}))
29

30
        # Calculate overall score
31
        results['overall_score'] = self.calculate_security_score(results['validations'])
32

33
        return results
34

35
    def validate_jailer_config(self, jailer_config: Dict) -> Dict:
36
        """Validate jailer configuration"""
37

38
        validation = {
39
            'name': 'Jailer Configuration',
40
            'checks': [],
41
            'passed': 0,
42
            'total': 0
43
        }
44

45
        # Check if jailer is enabled
46
        check = {
47
            'name': 'Jailer enabled',
48
            'passed': jailer_config.get('enabled', False),
49
            'message': 'Jailer should be enabled for production'
50
        }
51
        validation['checks'].append(check)
52
        validation['total'] += 1
53
        if check['passed']:
54
            validation['passed'] += 1
55

56
        # Check UID/GID
57
        uid = jailer_config.get('uid', 0)
58
        check = {
59
            'name': 'Non-root UID',
60
            'passed': uid > 0,
61
            'message': f'UID should be > 0 (current: {uid})'
62
        }
63
        validation['checks'].append(check)
64
        validation['total'] += 1
65
        if check['passed']:
66
            validation['passed'] += 1
67

68
        # Check chroot
69
        chroot_enabled = jailer_config.get('chroot_base_dir') is not None
70
        check = {
71
            'name': 'Chroot enabled',
72
            'passed': chroot_enabled,
73
            'message': 'Chroot should be configured'
74
        }
75
        validation['checks'].append(check)
76
        validation['total'] += 1
77
        if check['passed']:
78
            validation['passed'] += 1
79

80
        # Check new PID namespace
81
        new_pid_ns = jailer_config.get('new_pid_ns', False)
82
        check = {
83
            'name': 'New PID namespace',
84
            'passed': new_pid_ns,
85
            'message': 'New PID namespace should be enabled'
86
        }
87
        validation['checks'].append(check)
88
        validation['total'] += 1
89
        if check['passed']:
90
            validation['passed'] += 1
91

92
        return validation
93

94
    def validate_seccomp_config(self, seccomp_config: Dict) -> Dict:
95
        """Validate seccomp configuration"""
96

97
        validation = {
98
            'name': 'Seccomp Configuration',
99
            'checks': [],
100
            'passed': 0,
101
            'total': 0
102
        }
103

104
        # Check if seccomp is enabled
105
        filters = seccomp_config.get('filters', [])
106
        check = {
107
            'name': 'Seccomp filters configured',
108
            'passed': len(filters) > 0,
109
            'message': 'Seccomp filters should be configured'
110
        }
111
        validation['checks'].append(check)
112
        validation['total'] += 1
113
        if check['passed']:
114
            validation['passed'] += 1
115

116
        # Check default action
117
        default_action = seccomp_config.get('default_action', '')
118
        check = {
119
            'name': 'Secure default action',
120
            'passed': default_action in ['kill_process', 'trap'],
121
            'message': f'Default action should be restrictive (current: {default_action})'
122
        }
123
        validation['checks'].append(check)
124
        validation['total'] += 1
125
        if check['passed']:
126
            validation['passed'] += 1
127

128
        # Check for dangerous syscalls
129
        dangerous_syscalls = ['ptrace', 'kexec_load', 'create_module', 'init_module']
130
        all_allowed_syscalls = []
131
        for filter_entry in filters:
132
            if filter_entry.get('action') == 'allow':
133
                all_allowed_syscalls.extend(filter_entry.get('syscalls', []))
134

135
        dangerous_found = [sc for sc in dangerous_syscalls if sc in all_allowed_syscalls]
136
        check = {
137
            'name': 'No dangerous syscalls allowed',
138
            'passed': len(dangerous_found) == 0,
139
            'message': f'Dangerous syscalls found: {dangerous_found}' if dangerous_found else 'No dangerous syscalls'
140
        }
141
        validation['checks'].append(check)
142
        validation['total'] += 1
143
        if check['passed']:
144
            validation['passed'] += 1
145

146
        return validation
147

148
    def validate_cgroup_config(self, cgroup_config: Dict) -> Dict:
149
        """Validate cgroup configuration"""
150

151
        validation = {
152
            'name': 'Cgroup Configuration',
153
            'checks': [],
154
            'passed': 0,
155
            'total': 0
156
        }
157

158
        # Check memory limit
159
        memory_mb = cgroup_config.get('memory_mb', 0)
160
        check = {
161
            'name': 'Memory limit configured',
162
            'passed': memory_mb > 0,
163
            'message': f'Memory limit should be > 0 (current: {memory_mb}MB)'
164
        }
165
        validation['checks'].append(check)
166
        validation['total'] += 1
167
        if check['passed']:
168
            validation['passed'] += 1
169

170
        # Check CPU limit
171
        cpu_quota = cgroup_config.get('cpu_quota', 0)
172
        check = {
173
            'name': 'CPU limit configured',
174
            'passed': cpu_quota > 0,
175
            'message': f'CPU quota should be > 0 (current: {cpu_quota})'
176
        }
177
        validation['checks'].append(check)
178
        validation['total'] += 1
179
        if check['passed']:
180
            validation['passed'] += 1
181

182
        # Check PID limit
183
        max_pids = cgroup_config.get('max_pids', 0)
184
        check = {
185
            'name': 'PID limit configured',
186
            'passed': max_pids > 0 and max_pids < 1000,
187
            'message': f'PID limit should be reasonable (current: {max_pids})'
188
        }
189
        validation['checks'].append(check)
190
        validation['total'] += 1
191
        if check['passed']:
192
            validation['passed'] += 1
193

194
        return validation
195

196
    def validate_network_security(self, network_config: Dict) -> Dict:
197
        """Validate network security configuration"""
198

199
        validation = {
200
            'name': 'Network Security',
201
            'checks': [],
202
            'passed': 0,
203
            'total': 0
204
        }
205

206
        # Check network isolation
207
        isolated = network_config.get('isolated', False)
208
        check = {
209
            'name': 'Network isolation enabled',
210
            'passed': isolated,
211
            'message': 'Network isolation should be enabled'
212
        }
213
        validation['checks'].append(check)
214
        validation['total'] += 1
215
        if check['passed']:
216
            validation['passed'] += 1
217

218
        # Check firewall rules
219
        firewall_rules = network_config.get('firewall_rules', [])
220
        check = {
221
            'name': 'Firewall rules configured',
222
            'passed': len(firewall_rules) > 0,
223
            'message': 'Firewall rules should be configured'
224
        }
225
        validation['checks'].append(check)
226
        validation['total'] += 1
227
        if check['passed']:
228
            validation['passed'] += 1
229

230
        # Check for default deny policy
231
        has_deny_all = any(
232
            rule.get('action') == 'DROP' and rule.get('source_cidr') == '0.0.0.0/0'
233
            for rule in firewall_rules
234
        )
235
        check = {
236
            'name': 'Default deny policy',
237
            'passed': has_deny_all,
238
            'message': 'Should have default deny-all rule'
239
        }
240
        validation['checks'].append(check)
241
        validation['total'] += 1
242
        if check['passed']:
243
            validation['passed'] += 1
244

245
        return validation
246

247
    def validate_resource_limits(self, resource_config: Dict) -> Dict:
248
        """Validate resource limit configuration"""
249

250
        validation = {
251
            'name': 'Resource Limits',
252
            'checks': [],
253
            'passed': 0,
254
            'total': 0
255
        }
256

257
        # Check memory limit is reasonable
258
        memory_mb = resource_config.get('memory_mb', 0)
259
        reasonable_memory = 64 <= memory_mb <= 4096  # 64MB to 4GB
260
        check = {
261
            'name': 'Reasonable memory limit',
262
            'passed': reasonable_memory,
263
            'message': f'Memory should be 64-4096MB (current: {memory_mb}MB)'
264
        }
265
        validation['checks'].append(check)
266
        validation['total'] += 1
267
        if check['passed']:
268
            validation['passed'] += 1
269

270
        # Check vCPU count
271
        vcpus = resource_config.get('vcpus', 0)
272
        reasonable_vcpus = 1 <= vcpus <= 8
273
        check = {
274
            'name': 'Reasonable vCPU count',
275
            'passed': reasonable_vcpus,
276
            'message': f'vCPUs should be 1-8 (current: {vcpus})'
277
        }
278
        validation['checks'].append(check)
279
        validation['total'] += 1
280
        if check['passed']:
281
            validation['passed'] += 1
282

283
        # Check disk size limit
284
        disk_mb = resource_config.get('disk_mb', 0)
285
        reasonable_disk = 100 <= disk_mb <= 10240  # 100MB to 10GB
286
        check = {
287
            'name': 'Reasonable disk limit',
288
            'passed': reasonable_disk,
289
            'message': f'Disk should be 100-10240MB (current: {disk_mb}MB)'
290
        }
291
        validation['checks'].append(check)
292
        validation['total'] += 1
293
        if check['passed']:
294
            validation['passed'] += 1
295

296
        return validation
297

298
    def calculate_security_score(self, validations: Dict) -> float:
299
        """Calculate overall security score"""
300

301
        total_checks = 0
302
        total_passed = 0
303

304
        for validation in validations.values():
305
            total_checks += validation['total']
306
            total_passed += validation['passed']
307

308
        if total_checks == 0:
309
            return 0.0
310

311
        return (total_passed / total_checks) * 100
312

313
    def generate_report(self, validation_results: Dict) -> str:
314
        """Generate human-readable security report"""
315

316
        report = []
317
        report.append(f"Security Validation Report for VM: {validation_results['vm_id']}")
318
        report.append("=" * 50)
319
        report.append(f"Overall Security Score: {validation_results['overall_score']:.1f}%")
320
        report.append("")
321

322
        for validation in validation_results['validations'].values():
323
            report.append(f"{validation['name']}:")
324
            report.append(f"  Passed: {validation['passed']}/{validation['total']} checks")
325

326
            for check in validation['checks']:
327
                status = "✓" if check['passed'] else "✗"
328
                report.append(f"    {status} {check['name']}: {check['message']}")
329

330
            report.append("")
331

332
        # Recommendations
333
        report.append("Recommendations:")
334
        for validation in validation_results['validations'].values():
335
            for check in validation['checks']:
336
                if not check['passed']:
337
                    report.append(f"  • Fix: {check['message']}")
338

339
        return "\n".join(report)
340

341
# Usage example
342
if __name__ == '__main__':
343
    validator = SecurityValidator()
344

345
    # Example configuration
346
    config = {
347
        'jailer': {
348
            'enabled': True,
349
            'uid': 1000,
350
            'gid': 1000,
351
            'chroot_base_dir': '/srv/jailer',
352
            'new_pid_ns': True
353
        },
354
        'seccomp': {
355
            'default_action': 'kill_process',
356
            'filters': [
357
                {
358
                    'syscalls': ['read', 'write', 'open', 'close'],
359
                    'action': 'allow'
360
                }
361
            ]
362
        },
363
        'cgroups': {
364
            'memory_mb': 512,
365
            'cpu_quota': 50000,
366
            'max_pids': 100
367
        },
368
        'network': {
369
            'isolated': True,
370
            'firewall_rules': [
371
                {'action': 'ACCEPT', 'port': 22, 'source_cidr': '192.168.1.0/24'},
372
                {'action': 'DROP', 'source_cidr': '0.0.0.0/0'}
373
            ]
374
        },
375
        'resources': {
376
            'memory_mb': 512,
377
            'vcpus': 2,
378
            'disk_mb': 1024
379
        }
380
    }
381

382
    # Run validation
383
    results = validator.validate_all('vm001', config)
384

385
    # Generate report
386
    report = validator.generate_report(results)
387
    print(report)
388

389
    # Save results
390
    with open('security_validation.json', 'w') as f:
391
        json.dump(results, f, indent=2)

Conclusion#

Firecracker’s comprehensive security architecture provides multiple layers of protection, making it suitable for running untrusted workloads in multi-tenant environments. Key security features include:

🛡️ Hardware-assisted virtualization with KVM
🔒 Process isolation through jailer and Linux security primitives
🚫 Syscall filtering with seccomp-BPF
📊 Resource isolation and monitoring with cgroups
🌐 Network security and isolation
📋 Comprehensive auditing and monitoring capabilities

By implementing these security measures and following best practices, Firecracker provides enterprise-grade security for serverless computing platforms while maintaining exceptional performance and efficiency.