feat: Adding a new cli tool. Expanding the UI elements to be more intuitive. Rebuilt the system call diff logic so that it shows which ones are most dangerous

Author: antitree

Dockerfile

@@ -35,4 +35,4 @@ EXPOSE 5000
 RUN apt-get purge -y gcc && apt-get autoremove -y && rm -rf /var/lib/apt/lists/*
 
 # Command to run the Flask app
-CMD ["flask", "run"]
+CMD ["flask", "run", "--debug"]

common/diff.py

@@ -4,6 +4,42 @@
 from rich.console import Console
 from rich import box
 from rich.text import Text
+from rich.style import Style
+
+
+DANGEROUS_SYSCALLS = ["acct", "add_key", "bpf", "clock_adjtime", "clone", "create_module",
+        "delete_module", "finit_module", "get_kernel_syms", "get_mempolicy",
+        "init_module", "ioperm", "iopl", "kcmp", "kexec_file_load", "kexec_load",
+        "keyctl", "lookup_dcookie", "mbind", "mount", "move_pages", "nfsservctl",
+        "open_by_handle_at", "perf_event_open", "personality", "pivot_root",
+        "process_vm_readv", "process_vm_writev", "ptrace", "query_module",
+        "quotactl", "reboot", "request_key", "set_mempolicy", "setns",
+        "settimeofday", "stime", "swapon", "swapoff", "sysfs", "_sysctl", "umount",
+        "umount2", "unshare", "uselib", "userfaultfd", "ustat", "vm86", "vm86old"]
+
+def reduce_action(action):
+    ALLOW = ["ALLOW", action, "permissive"]
+    DENY = ["DENY", action, "restrictive"]
+    CONDITION = ["CONDITION", action, "restrictive"]
+    UNKNOWN = ["ERROR-NOT-FOUND", action, "permissive"]
+    ACTION_MAP = {
+        "ALLOW": ALLOW,
+        "ERRNO": DENY, 
+        "ALLOW/ERRORNO": CONDITION,
+        "N/A": ALLOW,
+        "LOG": ALLOW,
+        "KILL": DENY,
+        "TRACE": CONDITION,
+        "TRAP": CONDITION,
+        "Unknown": UNKNOWN,
+    }
+    # Check error numbers
+    if action.startswith("ERRNO"):
+        return DENY
+    elif "/" in action and action != "N/A":
+        return CONDITION
+    return ACTION_MAP.get(action, f"ERROR MAPPING EFFECTIVE PERMISSIONS {action}")
+    
 
 def is_convertible_to_int(s):
     """Check if a string can be safely converted to an integer."""
@@ -13,66 +49,51 @@ def is_convertible_to_int(s):
     except ValueError:
         return False
 
-dangerous_syscalls = [
-    "acct", "add_key", "bpf", "clock_adjtime", "clone", "create_module",
-    "delete_module", "finit_module", "get_kernel_syms", "get_mempolicy",
-    "init_module", "ioperm", "iopl", "kcmp", "kexec_file_load", "kexec_load",
-    "keyctl", "lookup_dcookie", "mbind", "mount", "move_pages", "nfsservctl",
-    "open_by_handle_at", "perf_event_open", "personality", "pivot_root",
-    "process_vm_readv", "process_vm_writev", "ptrace", "query_module",
-    "quotactl", "reboot", "request_key", "set_mempolicy", "setns",
-    "settimeofday", "stime", "swapon", "swapoff", "sysfs", "_sysctl", "umount",
-    "umount2", "unshare", "uselib", "userfaultfd", "ustat", "vm86", "vm86old"
-]
-
-def get_seccomp_policy(container1):
-    full1, d1 = get_seccomp_filters(container1["pid"])
-    if d1: 
-        container1["summary"] = d1.syscallSummary
-    else:
-        container1["summary"] = {}
-
-    da1 = d1.defaultAction
-
-    console = Console()
-    table = Table(show_header=True, show_lines=True, box=box.HEAVY_EDGE, style="green", pad_edge=False)
-    table.add_column(header="Container", justify="left", min_width=20)
-    table.add_column(header=container1["name"], justify="left", max_width=20, overflow=None)
-    table.add_column(header=container2["name"], justify="left", max_width=20, overflow=None)
-
-    # Add Seccomp and Capabilities Information
-    table.add_custom_row("[b]seccomp", container1["seccomp"])
-    table.add_custom_row("[b]caps", container1["caps"], end_section=True)
-
-    # Iterate through the global SYSCALLS dict
-    for syscall_num, syscall_info in SYSCALLS.items():
-        syscall_name = syscall_info[1]
-
-        # Determine effective policy for container1
-        if syscall_name in container1["summary"]:
-            action1 = container1["summary"][syscall_name].get("action", da1)
-            count1 = container1["summary"][syscall_name].get("count", 0)
-            effective_policy1 = f"{action1}"
+def get_seccomp_policy(container):
+    """Get the seccomp policy and return a detailed table for every syscall."""
+    try:
+        full, d = get_seccomp_filters(container["pid"])
+        if d:
+            container["summary"] = d.syscallSummary
         else:
-            effective_policy1 = f"{da1}"
+            container["summary"] = {}
 
-        table.add_custom_row(syscall_name, effective_policy1)
+        default_action = d.defaultAction if d else "unknown"
 
-    # Add total instructions row
-    container1["total"] = container1["summary"].get("total", {"count": 0}).get("count")
-    table.add_custom_row("Total Instructions", str(container1["total"]))
+        console = Console()
+        table = Table(show_header=True, show_lines=True, box=box.HEAVY_EDGE, style="green", pad_edge=False)
+        table.add_column(header="Syscall", justify="left", min_width=20)
+        table.add_column(header=f"{container['name']} Action", justify="left", min_width=20)
+
+        # Iterate through all syscalls in SYSCALLS
+        for syscall_num, syscall_info in SYSCALLS.items():
+            syscall_name = syscall_info[1]
+
+            # Get the action for this syscall from the container's summary, or default to the default action
+            if syscall_name in container["summary"]:
+                action = container["summary"][syscall_name].get("action", default_action)
+            else:
+                action = default_action
+
+            table.add_custom_row(syscall_name, action)
+
+        return table, full
+
+    except Exception as e:
+        console.print(f"An error occurred: {e}", style="bold red")
+        return None, None
+
+def compare_seccomp_policies(container1, container2, reduce=True, only_diff=True, only_dangerous=False):
+    """Compare the seccomp policies of two containers and return a detailed table."""
 
-    return table, full1
+    danger_style = Style(color="red", blink=True, bold=True)
 
-def compare_seccomp_policies(container1, container2, full=False):
+    
     try:
-        # Extract SeccompSummary for both PIDs
         full1, d1 = get_seccomp_filters(container1["pid"])
         full2, d2 = get_seccomp_filters(container2["pid"])
-        
-        
 
-        if d1: 
+        if d1:
             container1["summary"] = d1.syscallSummary
         else:
             container1["summary"] = {}
@@ -82,54 +103,69 @@ def compare_seccomp_policies(container1, container2, full=False):
         else:
             container2["summary"] = {}
 
-        da1 = d1.defaultAction
-        da2 = d2.defaultAction
+        default_action1 = d1.defaultAction if d1 else "unknown"
+        default_action2 = d2.defaultAction if d2 else "unknown"
 
         console = Console()
         table = Table(show_header=True, show_lines=True, box=box.HEAVY_EDGE, style="green", pad_edge=False)
-        table.add_column(header="Container", justify="left", min_width=20)
-        table.add_column(header=container1["name"], justify="left", max_width=20, overflow=None)
-        table.add_column(header=container2["name"], justify="left", max_width=20, overflow=None)
-
+        table.add_column(header="Syscall", justify="left", min_width=20)
+        table.add_column(header=f"{container1['name']} Action", justify="left", min_width=20)
+        table.add_column(header=f"{container2['name']} Action", justify="left", min_width=20)
+        
         # Add Seccomp and Capabilities Information
-        table.add_custom_row("[b]seccomp", container1["seccomp"], container2["seccomp"])
-        table.add_custom_row("[b]caps", container1["caps"], container2["caps"], end_section=True)
+        table.add_custom_row("[b]seccomp", container1["seccomp"])
+        table.add_custom_row("[b]caps", container1["caps"], end_section=True)
+        
 
-        # Iterate through the global SYSCALLS dict
+        # Iterate through all syscalls in SYSCALLS
         for syscall_num, syscall_info in SYSCALLS.items():
             syscall_name = syscall_info[1]
+            
+            if only_dangerous and not syscall_name in DANGEROUS_SYSCALLS:
+                continue
+            
+            
 
-            # Determine effective policy for container1
+            # Get the action for container1
             if syscall_name in container1["summary"]:
-                action1 = container1["summary"][syscall_name].get("action", da1)
-                count1 = container1["summary"][syscall_name].get("count", 0)
-                effective_policy1 = f"{action1}"
+                action1 = container1["summary"][syscall_name].get("action", default_action1)
             else:
-                effective_policy1 = f"{da1}"
+                action1 = default_action1
 
-            # Determine effective policy for container2
+            # Get the action for container2
             if syscall_name in container2["summary"]:
-                action2 = container2["summary"][syscall_name].get("action", da2)
-                count2 = container2["summary"][syscall_name].get("count", 0)
-                effective_policy2 = f"{action2}"
+                action2 = container2["summary"][syscall_name].get("action", default_action2)
             else:
-                effective_policy2 = f"{da2}"
-
-            # Compare effective policies
-            if effective_policy1 == effective_policy2:
-                continue  # Skip identical policies
-
-            table.add_custom_row(syscall_name, effective_policy1, effective_policy2)
-
-        # Add total instructions row
-        container1["total"] = container1["summary"].get("total", {"count": 0}).get("count")
-        container2["total"] = container2["summary"].get("total", {"count": 0}).get("count")
-        table.add_custom_row("Total Instructions", str(container1["total"]), str(container2["total"]))
-
-        if len(table.rows) <= 3 and container1["total"] == container2["total"]:
-            console.print(Text("No seccomp filter differences were found between the two containers", justify="center"))
+                action2 = default_action2
+                    
+            # Reduce the action to an effecctive action of allow or deny
+            if reduce:
+                action1 = reduce_action(action1)[0]
+                action2 = reduce_action(action2)[0]
+
+            # Skip identical policies if only_diff is True
+            if only_diff and action1 == action2:
+                continue
+            
+            if syscall_name in DANGEROUS_SYSCALLS:
+                syscall_name = f":warning:{syscall_name}"
+            
+            # Add row to table
+            table.add_custom_row(syscall_name, action1, action2)
 
-        return table, full1, full2
+    except Exception as e:
+        console.print(f"An error occurred: {e}", style="bold red")
+        return None, None, None
+            
+    # Add total instructions row
+    container1["total"] = container1["summary"].get("total", {"count": 0}).get("count")
+    container2["total"] = container2["summary"].get("total", {"count": 0}).get("count")
+    table.add_custom_row("Total Instructions", str(container1["total"]), str(container2["total"]))
+    if len(table.rows) <= 3 and container1["total"] == container2["total"]:
+        console.print(Text("No seccomp filter differences were found between the two containers", justify="center"))
+
+    return table, full1, full2
 
-    except ValueError as e:
-        print(f"An error occurred: {e}")
+    
+    
+    

common/ptrace.py

@@ -4,7 +4,7 @@
 import sys
 import ctypes
 import ctypes.util
-import argparse
+
 from rich.console import Console
 from rich.table import Table
 from rich import box
@@ -131,17 +131,34 @@ def list_seccomp_filters(pid, dump=False, summary=True, allarch=True):
 
     if allarch:
         print("Filtering for non-x86-specific instructions is not implemented yet.")
+        
 
-def main():
-    """Entry point for the script."""
-    parser = argparse.ArgumentParser(description="Inspect seccomp profiles for a given PID.")
-    parser.add_argument("pid", type=int, help="PID of the process to inspect")
-    parser.add_argument("--dump", action="store_true", help="Dump the raw seccomp filters")
-    parser.add_argument("--summary", action="store_true", help="Display a summary of the seccomp filters")
-    parser.add_argument("--allarch", action="store_true", help="Search for all syscalls across any architecture")
-
-    args = parser.parse_args()
-    list_seccomp_filters(args.pid, dump=args.dump, summary=args.summary, allarch=args.allarch)
 
-if __name__ == "__main__":
-    main()
+def list_seccomp_pids():
+    # Iterate through all the folders in /proc
+    FILTER = True
+    
+    console = Console()
+    table = Table(show_header=False, show_lines=True, box=box.HEAVY_EDGE, style="green", pad_edge=False)
+    table.add_column(header="PID", justify="left", min_width=20)
+    table.add_column(header="Seccomp Mode", justify="left", min_width=20)
+
+    for pid in os.listdir('/proc'):
+        if pid.isdigit():
+            status_path = f"/proc/{pid}/status"
+            # Check if the status file exists for this PID
+            cmd = "Unknown"
+            if os.path.isfile(status_path):
+                with open(status_path, 'r') as status_file:
+                    for line in status_file:
+                        if line.startswith("Name:"):
+                            cmd = line.split()[1]
+                        if line.startswith("Seccomp_filters:"):
+                            no_instructions = line.split()[1]
+                            if FILTER and int(no_instructions) < 1:
+                                break
+                            else: 
+                                # Print the PID and the seccomp value
+                                table.add_row(f"{cmd}({pid})",no_instructions)
+                            break
+    return table

seccomp_dump.py

@@ -0,0 +1,32 @@
+import common.ptrace as ptrace
+import argparse
+from rich.console import Console
+
+
+def main():
+    """Entry point for the script."""
+    parser = argparse.ArgumentParser(description="Inspect seccomp profiles for a given PID.")
+    parser.add_argument("pid", type=int, nargs="?", help="PID of the process to inspect")
+    parser.add_argument("--dump", action="store_true", help="Dump the raw seccomp filters")
+    parser.add_argument("--summary", action="store_true", help="Display a summary of the seccomp filters")
+    parser.add_argument("--list", action="store_true", help="Display a list of pids with seccomp filters")
+    parser.add_argument("--allarch", action="store_true", help="Search for all syscalls across any architecture")
+
+    args = parser.parse_args()
+    
+    if not args.list and args.pid is None:
+        parser.error("PID is required unless --list is specified.")
+        
+     # If PID is provided and no other options are set, assume --dump
+    if args.pid is not None and not (args.dump or args.summary or args.list or args.allarch):
+        args.dump = True
+    
+    if args.list:
+        table = ptrace.list_seccomp_pids()
+        c = Console()
+        c.print(table)
+    else: 
+        ptrace.list_seccomp_filters(args.pid, dump=args.dump, summary=args.summary, allarch=args.allarch)
+
+if __name__ == "__main__":
+    main()