سياسات الأمان لتنفيذ التعليمات البرمجية

لماذا لا تكفي السياسات وحدها

نظرة عامة على قيود القدرات

import docker
client = docker.from_env()

# Drop all Linux capabilities
output = client.containers.run(
    'python:3.12-slim',
    'python -c "print(\"safe run\")"',
    cap_drop=['ALL'],
    network_disabled=True,
    mem_limit='256m',
    auto_remove=True
)
print(output.decode())

حظر الكتابة على القرص خارج /tmp

import tempfile, docker
client = docker.from_env()

with tempfile.TemporaryDirectory() as tmpdir:
    output = client.containers.run(
        'python:3.12-slim',
        'python -c "open(\"/tmp/ok.txt\",\"w\").write(\"ok\"); print(open(\"/tmp/ok.txt\").read())"',
        read_only=True,
        tmpfs={'/tmp': 'size=64m,mode=1777'},
        network_disabled=True,
        mem_limit='128m',
        auto_remove=True
    )
    print(output.decode())  # 'ok' — /tmp write allowed
    # Writing anywhere else raises ReadOnlyFileSystem

ملفات تعريف seccomp: تصفية استدعاءات النظام

import json, docker
client = docker.from_env()

# Minimal seccomp profile: only allow read, write, open, close, exit
seccomp_profile = json.dumps({
    'defaultAction': 'SCMP_ACT_ERRNO',
    'architectures': ['SCMP_ARCH_X86_64'],
    'syscalls': [
        {'names': ['read', 'write', 'open', 'openat', 'close',
                   'fstat', 'mmap', 'mprotect', 'munmap', 'brk',
                   'exit', 'exit_group', 'rt_sigaction',
                   'rt_sigprocmask', 'futex'],
         'action': 'SCMP_ACT_ALLOW'}
    ]
})

output = client.containers.run(
    'python:3.12-slim',
    'python -c "print(1+1)"',
    security_opt=[f'seccomp={seccomp_profile}'],
    auto_remove=True
)
print(output.decode())

حدود الموارد: وحدة المعالجة المركزية والذاكرة والوقت

output = client.containers.run(
    'python:3.12-slim',
    'python -c "import time; time.sleep(100)"',
    mem_limit='128m',
    nano_cpus=500_000_000,   # 0.5 CPU
    network_disabled=True,
    auto_remove=True,
    timeout=10               # kill after 10 seconds
)
# Raises ReadTimeout after 10 seconds

السماح بقائمة استيرادات Python

import ast

ALLOWED_IMPORTS = {'math', 'statistics', 'json', 'csv', 'datetime', 'collections', 're'}

def check_imports(code: str) -> list[str]:
    tree = ast.parse(code)
    blocked = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name.split('.')[0] not in ALLOWED_IMPORTS:
                    blocked.append(alias.name)
        elif isinstance(node, ast.ImportFrom):
            if node.module and node.module.split('.')[0] not in ALLOWED_IMPORTS:
                blocked.append(node.module)
    return blocked

print(check_imports('import socket; print(1)'))   # ['socket']
print(check_imports('import math; print(math.pi)'))  # []

اكتشاف أنماط التعليمات البرمجية الخطرة

import ast

DANGEROUS_CALLS = {'exec', 'eval', 'compile', '__import__', 'breakpoint'}

def check_dangerous_calls(code: str) -> list[str]:
    tree = ast.parse(code)
    found = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            if isinstance(node.func, ast.Name):
                if node.func.id in DANGEROUS_CALLS:
                    found.append(node.func.id)
            elif isinstance(node.func, ast.Attribute):
                if node.func.attr in DANGEROUS_CALLS:
                    found.append(node.func.attr)
    return found

print(check_dangerous_calls('exec("import os")'))  # ['exec']

حدود حجم المخرجات

MAX_OUTPUT_BYTES = 1 * 1024 * 1024  # 1 MB

def run_with_output_limit(client, code: str) -> dict:
    container = client.containers.run(
        'python:3.12-slim',
        f'python -c "{code}"',
        mem_limit='256m',
        network_disabled=True,
        detach=True
    )
    try:
        container.wait(timeout=30)
        logs = container.logs(stdout=True, stderr=True)
        if len(logs) > MAX_OUTPUT_BYTES:
            return {'success': False, 'error': 'Output too large'}
        return {'success': True, 'output': logs.decode(errors='replace')}
    finally:
        container.remove(force=True)

if __name__ == '__main__':
    class FakeContainer:
        def wait(self, timeout=30): pass
        def logs(self, stdout=True, stderr=True): return b'Analysis complete: 42 rows processed.'
        def remove(self, force=True): pass
    class FakeContainers:
        def run(self, *args, **kwargs): return FakeContainer()
    class FakeClient:
        containers = FakeContainers()

    result = run_with_output_limit(FakeClient(), 'print(42)')
    print('Success:', result['success'])
    print('Output :', result['output'])

علامة no-new-privileges

output = client.containers.run(
    'python:3.12-slim',
    'python -c "import os; print(os.getuid())"',
    user='nobody',                              # run as non-root
    security_opt=['no-new-privileges:true'],    # no privilege escalation
    cap_drop=['ALL'],                           # no capabilities
    network_disabled=True,
    mem_limit='128m',
    auto_remove=True
)
print(output.decode())  # numeric UID of 'nobody'

دمج جميع السياسات

import tempfile, json, docker
client = docker.from_env()

def maximum_security_run(code: str) -> dict:
    with tempfile.TemporaryDirectory() as tmpdir:
        try:
            out = client.containers.run(
                'python:3.12-slim',
                f'python -c "{code}"',
                user='nobody',
                security_opt=['no-new-privileges:true'],
                cap_drop=['ALL'],
                network_disabled=True,
                read_only=True,
                tmpfs={'/tmp': 'size=32m,mode=1777'},
                mem_limit='128m',
                nano_cpus=500_000_000,
                auto_remove=True,
                timeout=15
            )
            return {'success': True, 'output': out.decode()}
        except Exception as e:
            return {'success': False, 'error': str(e)}

تسجيل انتهاكات السياسات

import hashlib, logging, time

logging.basicConfig(level=logging.INFO)
security_logger = logging.getLogger('sandbox.security')

def policy_check_and_run(code: str, session_id: str) -> dict:
    violations = check_imports(code) + check_dangerous_calls(code)
    code_hash = hashlib.sha256(code.encode()).hexdigest()[:16]

    if violations:
        security_logger.warning(
            'POLICY_VIOLATION session=%s hash=%s violations=%s',
            session_id, code_hash, violations
        )
        return {'success': False, 'error': f'Blocked: {violations}'}

    result = maximum_security_run(code)
    security_logger.info(
        'EXEC session=%s hash=%s success=%s',
        session_id, code_hash, result['success']
    )
    return result

ما وظيفة الإعداد 'defaultAction: SCMP_ACT_ERRNO' في ملف تعريف seccomp؟

مراجعة سياسات الأمان