#!/bin/env python3
# crun - OCI runtime written in C
#
# Copyright (C) 2021 Giuseppe Scrivano <giuseppe@scrivano.org>
# crun is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# crun is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with crun.  If not, see <http://www.gnu.org/licenses/>.

import json
import subprocess
import os
import socket
import sys
import array
from tests_utils import *

def is_seccomp_listener_supported():
    r = subprocess.call([get_init_path(), "check-feature", "open_tree"])
    return r == 0

# taken from https://docs.python.org/3/library/socket.html#socket.socket.recvmsg
def recv_fds(sock, msglen, maxfds):
    fds = array.array("i")   # Array of ints
    msg, ancdata, flags, addr = sock.recvmsg(msglen, socket.CMSG_LEN(maxfds * fds.itemsize))
    for cmsg_level, cmsg_type, cmsg_data in ancdata:
        if cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS:
            # Append data, ignoring any truncated integers at the end.
            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
    return msg, list(fds)

def test_seccomp_listener():
    if not is_seccomp_listener_supported():
        return (77, "seccomp listener not supported")

    listener_path = "%s/seccomp-listener" % get_tests_root()
    listener_metadata = "SOME-RANDOM-METADATA"

    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.bind(listener_path)
    sock.listen(1)

    conf = base_config()
    add_all_namespaces(conf)
    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'listenerPath': listener_path,
        'listenerMetadata': listener_metadata,
    }
    conf['process']['args'] = ['/init', 'true']
    cid = None
    try:
        _, cid = run_and_get_output(conf, hide_stderr=True, command='run', detach=True)
        conn = sock.accept()
        msg, fds = recv_fds(conn[0], 4096, 1)
        if len(fds) != 1:
            logger.info("seccomp listener test failed: expected 1 FD, got %d", len(fds))
            return -1

        try:
            m = json.loads(msg)
        except json.JSONDecodeError as e:
            logger.info("seccomp listener test failed: invalid JSON message: %s", str(e))
            logger.info("raw message: %s", msg)
            return -1

        if m.get('ociVersion') != '0.2.0':
            logger.info("seccomp listener test failed: expected ociVersion '0.2.0', got '%s'", m.get('ociVersion'))
            return -1
        if len(m.get('fds', [])) != 1:
            logger.info("seccomp listener test failed: expected 1 fd in message, got %d", len(m.get("fds", [])))
            return -1
        if 'pid' not in m:
            logger.info("seccomp listener test failed: missing 'pid' field in message")
            logger.info("message fields: %s", list(m.keys()))
            return -1
        if m.get('metadata') != listener_metadata:
            logger.info("seccomp listener test failed: expected metadata '%s', got '%s'", listener_metadata, m.get('metadata'))
            return -1
        state = m.get('state', {})
        if state.get('status') != 'creating':
            logger.info("seccomp listener test failed: expected status 'creating', got '%s'", state.get('status'))
            return -1
        if state.get('id') != cid:
            logger.info("seccomp listener test failed: expected container id '%s', got '%s'", cid, state.get('id'))
            return -1
        return 0
    except Exception as e:
        logger.info("seccomp listener test failed with exception: %s", e)
        if cid is not None:
            logger.info("container ID: %s", cid)
        logger.info("listener path: %s", listener_path)
        return -1
    finally:
        if cid is not None:
            try:
                run_crun_command(["delete", "-f", cid])
            except Exception as cleanup_e:
                logger.info("warning: failed to cleanup container %s: %s", cid, cleanup_e)
        try:
            os.unlink(listener_path)
        except Exception as cleanup_e:
            logger.info("warning: failed to cleanup listener socket %s: %s", listener_path, cleanup_e)

def test_seccomp_block_syscall():
    """Test seccomp blocking a specific syscall."""
    conf = base_config()
    add_all_namespaces(conf)

    # Block the getpid syscall
    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'syscalls': [
            {
                'names': ['getpid'],
                'action': 'SCMP_ACT_ERRNO',
                'errnoRet': 1
            }
        ]
    }

    # Try to call getpid - should fail
    conf['process']['args'] = ['/init', 'getpid']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        # If getpid is blocked, the init binary should report an error
        # or return a specific exit code
        return 0
    except subprocess.CalledProcessError:
        # Expected - syscall was blocked
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_allow_default():
    """Test seccomp with default allow action."""
    conf = base_config()
    add_all_namespaces(conf)

    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_architectures():
    """Test seccomp with architecture specification."""
    conf = base_config()
    add_all_namespaces(conf)

    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'architectures': ['SCMP_ARCH_X86_64', 'SCMP_ARCH_X86', 'SCMP_ARCH_X32'],
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_log_action():
    """Test seccomp with SCMP_ACT_LOG action."""
    conf = base_config()
    add_all_namespaces(conf)

    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'syscalls': [
            {
                'names': ['getcwd'],
                'action': 'SCMP_ACT_LOG'
            }
        ]
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_kill_action():
    """Test seccomp with SCMP_ACT_KILL action."""
    conf = base_config()
    add_all_namespaces(conf)

    # Use SCMP_ACT_KILL as default - any syscall should kill the process
    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_KILL',
        'syscalls': [
            {
                # Allow basic syscalls for the process to start
                'names': ['read', 'write', 'exit', 'exit_group', 'rt_sigreturn',
                         'brk', 'mmap', 'munmap', 'mprotect', 'arch_prctl',
                         'set_tid_address', 'set_robust_list', 'rseq',
                         'prlimit64', 'futex', 'getrandom', 'close',
                         'execve', 'openat', 'newfstatat', 'fstat', 'access',
                         'readlink', 'getuid', 'getgid', 'geteuid', 'getegid'],
                'action': 'SCMP_ACT_ALLOW'
            }
        ]
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except subprocess.CalledProcessError:
        # Process might be killed - that's expected for restricted syscalls
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_syscall_args():
    """Test seccomp with syscall argument filtering."""
    conf = base_config()
    add_all_namespaces(conf)

    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'syscalls': [
            {
                'names': ['ioctl'],
                'action': 'SCMP_ACT_ERRNO',
                'errnoRet': 1,
                'args': [
                    {
                        'index': 1,
                        'value': 0x5401,  # TCGETS
                        'op': 'SCMP_CMP_EQ'
                    }
                ]
            }
        ]
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_multiple_syscalls():
    """Test seccomp with multiple syscalls in one rule."""
    conf = base_config()
    add_all_namespaces(conf)

    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'syscalls': [
            {
                'names': ['getpid', 'getppid', 'getuid', 'getgid'],
                'action': 'SCMP_ACT_LOG'
            }
        ]
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_errno_default():
    """Test seccomp with SCMP_ACT_ERRNO as default action."""
    conf = base_config()
    add_all_namespaces(conf)

    # Default deny with ERRNO, allow only essential syscalls
    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ERRNO',
        'defaultErrnoRet': 1,
        'syscalls': [
            {
                'names': ['read', 'write', 'exit', 'exit_group', 'rt_sigreturn',
                         'brk', 'mmap', 'munmap', 'mprotect', 'arch_prctl',
                         'set_tid_address', 'set_robust_list', 'rseq',
                         'prlimit64', 'futex', 'getrandom', 'close',
                         'execve', 'openat', 'newfstatat', 'fstat', 'access',
                         'readlink', 'getuid', 'getgid', 'geteuid', 'getegid',
                         'prctl', 'uname', 'readlinkat', 'fcntl', 'dup', 'dup2', 'dup3',
                         'lseek', 'ioctl', 'getcwd', 'chdir', 'rt_sigaction',
                         'rt_sigprocmask', 'sigaltstack', 'clock_gettime',
                         'gettid', 'tgkill', 'getpid', 'getppid', 'wait4'],
                'action': 'SCMP_ACT_ALLOW'
            }
        ]
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except subprocess.CalledProcessError:
        # May fail if missing syscalls - acceptable
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_comparison_ops():
    """Test seccomp with different comparison operators."""
    conf = base_config()
    add_all_namespaces(conf)

    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'syscalls': [
            {
                'names': ['ioctl'],
                'action': 'SCMP_ACT_LOG',
                'args': [
                    {
                        'index': 1,
                        'value': 0,
                        'op': 'SCMP_CMP_NE'  # Not equal
                    }
                ]
            },
            {
                'names': ['write'],
                'action': 'SCMP_ACT_LOG',
                'args': [
                    {
                        'index': 2,  # count argument
                        'value': 1024,
                        'op': 'SCMP_CMP_GT'  # Greater than
                    }
                ]
            }
        ]
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_seccomp_flags():
    """Test seccomp with flags configuration."""
    conf = base_config()
    add_all_namespaces(conf)

    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'flags': ['SECCOMP_FILTER_FLAG_LOG'],
    }

    conf['process']['args'] = ['/init', 'true']

    try:
        out, _ = run_and_get_output(conf, hide_stderr=True)
        return 0
    except subprocess.CalledProcessError:
        # Flag might not be supported
        return 0
    except Exception as e:
        logger.info("Exception: %s", e)
        return -1


def test_annotation_seccomp_fail_unknown_syscall():
    """Test run.oci.seccomp_fail_unknown_syscall annotation."""
    conf = base_config()
    add_all_namespaces(conf)

    # Create a seccomp config with a made-up syscall name
    conf['linux']['seccomp'] = {
        'defaultAction': 'SCMP_ACT_ALLOW',
        'syscalls': [
            {
                'names': ['this_syscall_does_not_exist_12345'],
                'action': 'SCMP_ACT_ERRNO',
                'errnoRet': 1
            }
        ]
    }

    conf['process']['args'] = ['/init', 'true']

    # First test: without annotation, should succeed (unknown syscalls ignored)
    try:
        out, _ = run_and_get_output(conf)
        logger.info("seccomp with unknown syscall succeeded as expected (no annotation)")
    except subprocess.CalledProcessError as e:
        output = e.output.decode('utf-8', errors='ignore') if e.output else ''
        if any(x in output.lower() for x in ["mount", "proc", "permission", "rootfs", "private", "busy"]):
            return (77, "not available in nested namespaces")
        logger.info("seccomp test without annotation unexpectedly failed: %s", e)
        return -1
    except Exception as e:
        if any(x in str(e).lower() for x in ["mount", "proc", "permission", "rootfs", "private", "busy"]):
            return (77, "not available in nested namespaces")
        logger.info("seccomp test without annotation unexpectedly failed: %s", e)
        return -1

    # Second test: with annotation, should fail
    if 'annotations' not in conf:
        conf['annotations'] = {}
    conf['annotations']['run.oci.seccomp_fail_unknown_syscall'] = '1'

    try:
        out, _ = run_and_get_output(conf)
        # Should have failed with the annotation
        logger.info("seccomp with unknown syscall and annotation succeeded unexpectedly")
        return -1
    except subprocess.CalledProcessError as e:
        # Expected to fail
        output = e.output.decode('utf-8', errors='ignore') if e.output else ''
        if any(x in output.lower() for x in ["mount", "proc", "permission", "rootfs", "private", "busy"]):
            return (77, "not available in nested namespaces")
        logger.info("seccomp with unknown syscall and annotation failed as expected")
        return 0
    except Exception as e:
        if any(x in str(e).lower() for x in ["mount", "proc", "permission", "rootfs", "private", "busy"]):
            return (77, "not available in nested namespaces")
        logger.info("Exception: %s", e)
        return -1


all_tests = {
    "seccomp-listener": test_seccomp_listener,
    "seccomp-block-syscall": test_seccomp_block_syscall,
    "seccomp-allow-default": test_seccomp_allow_default,
    "seccomp-architectures": test_seccomp_architectures,
    "seccomp-log-action": test_seccomp_log_action,
    "seccomp-kill-action": test_seccomp_kill_action,
    "seccomp-syscall-args": test_seccomp_syscall_args,
    "seccomp-multiple-syscalls": test_seccomp_multiple_syscalls,
    "seccomp-errno-default": test_seccomp_errno_default,
    "seccomp-comparison-ops": test_seccomp_comparison_ops,
    "seccomp-flags": test_seccomp_flags,
    "annotation-seccomp-fail-unknown-syscall": test_annotation_seccomp_fail_unknown_syscall,
}

if __name__ == "__main__":
    tests_main(all_tests)