labgrid-raw-interface: Add permissions

helpers/labgrid-raw-interface

@@ -11,19 +11,27 @@ import argparse
 import os
 import string
 import sys
+import json
 
 import yaml
 
+DENYLIST_FILE = "/etc/labgrid/helpers.yaml"
+PROGRAMS = ("tcpreplay", "tcpdump", "ip", "ethtool", "allowed-commands")
 
-def get_denylist():
-    denylist_file = "/etc/labgrid/helpers.yaml"
+
+def get_config():
     try:
-        with open(denylist_file) as stream:
+        with open(DENYLIST_FILE) as stream:
             data = yaml.load(stream, Loader=yaml.SafeLoader)
     except (PermissionError, FileNotFoundError, AttributeError) as e:
-        raise Exception(f"No configuration file ({denylist_file}), inaccessable or invalid yaml") from e
+        raise Exception(f"No configuration file ({DENYLIST_FILE}), inaccessable or invalid yaml") from e
+
+    return data.get("raw-interface", {})
+
 
-    denylist = data.get("raw-interface", {}).get("denied-interfaces", [])
+def get_denylist(config):
+    # FIXME: don't default to empty list for the check below
+    denylist = config.get("denied-interfaces", [])
 
     if not isinstance(denylist, list):
         raise Exception("No explicit denied-interfaces or not a list, please check your configuration")
@@ -33,26 +41,42 @@ def get_denylist():
     return denylist
 
 
+def get_allowed_commands(config):
+    allowed = set(config.get("allowed-commands", PROGRAMS))
+    allowed.add("allowed-commands")
+    return allowed
+
+
 def main(program, options):
+    config = get_config()
+
+    if program not in PROGRAMS:
+        raise ValueError(f"Invalid program {program} called with wrapper, valid programs are: {PROGRAMS}")
+
+    allowed_commands = get_allowed_commands(config)
+    if program not in allowed_commands:
+        raise ValueError(f"Command '{program}' is not allowed.")
+
+    args = [
+        program,
+    ]
+
+    if program == "allowed-commands":
+        print(json.dumps(sorted(list(allowed_commands))))
+        return 0
+
     if not options.ifname:
         raise ValueError("Empty interface name.")
     if any((c == "/" or c.isspace()) for c in options.ifname):
         raise ValueError(f"Interface name '{options.ifname}' contains invalid characters.")
     if len(options.ifname) > 15:
         raise ValueError(f"Interface name '{options.ifname}' is too long.")
 
-    denylist = get_denylist()
+    denylist = get_denylist(config)
 
     if options.ifname in denylist:
         raise ValueError(f"Interface name '{options.ifname}' is denied in denylist.")
 
-    programs = ["tcpreplay", "tcpdump", "ip", "ethtool"]
-    if program not in programs:
-        raise ValueError(f"Invalid program {program} called with wrapper, valid programs are: {programs}")
-
-    args = [
-        program,
-    ]
 
     if program == "tcpreplay":
         args.append(f"--intf1={options.ifname}")
@@ -123,6 +147,9 @@ if __name__ == "__main__":
     parser.add_argument("-d", "--debug", action="store_true", default=False, help="enable debug mode")
     subparsers = parser.add_subparsers(dest="program", help="program to run")
 
+    # allowed-commands
+    allowed_commands_parser = subparsers.add_parser("allowed-commands")
+
     # tcpdump
     tcpdump_parser = subparsers.add_parser("tcpdump")
     tcpdump_parser.add_argument("ifname", type=str, help="interface name")

labgrid/driver/rawnetworkinterfacedriver.py

@@ -35,14 +35,18 @@ def __attrs_post_init__(self):
         super().__attrs_post_init__()
         self._record_handle = None
         self._replay_handle = None
+        self._allowed_commands = []
 
     def on_activate(self):
-        if self.manage_interface:
+        self._allowed_commands = set(
+            json.loads(subprocess.run(self._wrap_command(["allowed-commands"]), stdout=subprocess.PIPE).stdout)
+        )
+        if self.manage_interface and "ip" in self._allowed_commands:
             self._set_interface("up")
             self._wait_state("up")
 
     def on_deactivate(self):
-        if self.manage_interface:
+        if self.manage_interface and "ip" in self._allowed_commands:
             self._set_interface("down")
             self._wait_state("down")