in progress

Signed-off-by: John Andersen <[email protected]>

Author: pdxjohnny

dffml/util/testing/manifest/shim.py

@@ -107,12 +107,11 @@
 -----
 
 - https://github.com/mjg59/ssh_pki
-
-    - Should we use this? No. Are we going to? Yes.
 """
 import os
 import sys
 import pathlib
+import argparse
 import importlib
 import contextlib
 import dataclasses
@@ -159,89 +158,41 @@ def parse(contents: str):
         except Exception as yaml_parse_error:
             raise yaml_parse_error from json_parse_error
 
-        from pprint import pprint
-
-        # Known parser mapping
-        parse = {
-            (
-                "tps.manifest",
-                "0.0.0",
-                "dataflow",
-            ): self.parse_my_document_format_0_0_0_dataflow
-        }
-        # Grab mapped parser
-        document_format_version_output_mode = (
-            doc.get("$document_format", None),
-            doc.get("$document_version", None),
-            doc.get("$document_version", None),
-        )
-        parser = parse.get(document_format_version, None)
-
-        if parser is None:
-            raise Exception(
-                f"Unknown document format/version pair: {document_format_version}"
-            )
-
-        print()
-        pprint(doc)
-        print()
-        parser(doc)
-
-    def parse_my_document_format_0_0_0_dataflow(self, doc):
-        pass
-
-
-@dataclasses.dataclass
-class ManifestFormatParser:
-    """
-    Read in configuration to determine what the next phase of parsing is.
-
-    args holds arguments passed to target.
-    """
-
-    format_name: str
-    version: str
-    output: str
-    action: str
-    target: str
-    args: str = ""
-
 
-ENV_PREFIX = "TPS_MANIFEST_"
-
-
-def environ_discover_dataclass(
+def discover_dataclass_environ(
     dataclass,
+    prefix: str,
     environ: Dict[str, str] = None,
     *,
-    prefix: str = ENV_PREFIX,
     dataclass_key: str = None,
 ):
     r"""
     >>> import dataclasses
-    >>> from dffml.util.testing.manifest.shim import environ_discover_dataclass
+    >>> from dffml.util.testing.manifest.shim import discover_dataclass_environ
     >>>
     >>> @dataclasses.dataclass
     ... class MyDataclass:
     ...     name: str
     ...     version: str
+    ...
+    ...     PREFIX = "MYPREFIX_"
     >>>
-    >>> environ_discover_dataclass(
+    >>> discover_dataclass_environ(
     ...     MyDataclass,
+    ...     MyDataclass.PREFIX,
     ...     {
     ...         "MYPREFIX_NAME_EXAMPLE_FORMAT": "Example Format",
     ...         "MYPREFIX_VERSION_EXAMPLE_FORMAT": "0.0.1",
     ...     },
-    ...     prefix="MYPREFIX_",
     ... )
     {'example_format': MyDataclass(name='Example Format', version='0.0.1')}
     >>>
-    >>> environ_discover_dataclass(
+    >>> discover_dataclass_environ(
     ...     MyDataclass,
+    ...     MyDataclass.PREFIX,
     ...     {
     ...         "MYPREFIX_VERSION_EXAMPLE_FORMAT": "0.0.1",
     ...     },
-    ...     prefix="MYPREFIX_",
     ...     dataclass_key="name",
     ... )
     {'example_format': MyDataclass(name='example_format', version='0.0.1')}
@@ -265,11 +216,144 @@ def environ_discover_dataclass(
     return discovered_parsers
 
 
-def shim(manifest: str, lockdown: bool, strict: bool):
-    parsers = environ_discover_dataclass(
-        ManifestFormatParser, dataclass_key="format_name", environ=os.environ
+@dataclasses.dataclass
+class ManifestFormatParser:
+    """
+    Read in configuration to determine what the next phase of parsing is.
+
+    args holds arguments passed to target.
+    """
+
+    format_name: str
+    version: str
+    output: str
+    action: str
+    target: str
+    args: str = ""
+
+    PREFIX: str = "TPS_MANIFEST_PARSER_"
+    DATACLASS_KEY: str = "format_name"
+
+
+def input_action_print(
+    args: argparse.Namespace, parser: ManifestFormatParser, manifest: Dict
+):
+    """
+    Print the formated version of the action.
+    """
+    return print()
+
+
+def input_action_exec(
+    args: argparse.Namespace, parser: ManifestFormatParser, manifest: Dict
+):
+    """
+    Execute the action via :py:func:`os.execvp`.
+    """
+    raise NotImplementedError()
+
+
+FORMAT_PARSER_ACTIONS = {
+    "print": input_action_print,
+    "exec": input_action_exec,
+}
+
+
+def input_action_stdin(args: argparse.Namespace):
+    """
+    Read manifest from stdin
+    """
+    return sys.stdin.read()
+
+
+def input_action_read_text(args: argparse.Namespace):
+    """
+    Read manifest from target filepath as text
+    """
+    if args.input_target is None:
+        raise ValueError("input target must be filepath but was None")
+    input_target_path = pathlib.Path(args.input_target)
+    if not input_target_path.exists():
+        raise ValueError(
+            f"input target must be filepath but was {args.input_target!r}"
+        )
+    return input_target_path.read_text()
+
+
+INPUT_ACTIONS = {
+    "stdin": input_action_stdin,
+    "read_text": input_action_read_text,
+}
+
+
+def shim(
+    args: argparse.Namespace, environ: Dict[str, str] = None,
+):
+    r"""
+
+    **TODO** Find code that sends all rest of args to target (QEMU?)
+
+    >>> import dataclasses
+    >>> import sys
+    >>> import types
+    >>> import pprint
+    >>> from dffml.util.testing.manifest.shim import shim
+    >>>
+    >>> @dataclasses.dataclass
+    ... class MyDataclass:
+    ...     name: str
+    ...     version: str
+    ...
+    ...     PREFIX = "MYPREFIX_"
+    >>>
+    >>> shim(
+    ...     types.SimpleNamespace(
+    ...         action,
+    ...     ),
+    ...     {
+    ...         "TPS_MANIFSET_ACTION_ONE": "print",
+    ...         "TPS_MANIFSET_ACTION_TWO": "print",
+    ...     }
+    ... )
+    """
+    # Discover options for format parsers for next phase
+    parsers = {
+        (parser.format_name, parser.version, parser.action)
+        for parser in discover_dataclass_environ(
+            ManifestFormatParser,
+            ManifestFormatParser.PREFIX,
+            dataclass_key=ManifestFormatParser.DATACLASS_KEY,
+            environ=os.environ,
+        )
+    }
+    # Determine how to get the manifest
+    input_action = INPUT_ACTIONS[args.input_action]
+    # Get the manifest
+    contents = input_action(args)
+    # Validate the manfiest
+    if not args.insecure:
+        # TODO Implement validation
+        pass
+    if args.only_validate:
+        # Bail if we are only validating the manfest and not parsing it
+        return
+    # Parse the manifest
+    manifest = parse(contents)
+    # Grab mapped parser
+    format_version_action = (
+        manifest.get("$document_format", None),
+        manifest.get("$document_version", None),
+        args.action,
     )
-    print(parsers)
+    if format_version_action not in parsers:
+        raise Exception(
+            f"Unknown document format/version/action combination: {format_version_action}: {parsers!r}"
+        )
+    parser = parsers[format_version_action]
+    # Determine how to get the manifest
+    action = FORMAT_PARSER_ACTIONS[parser.action]
+    # Send manifest to next phase
+    action(args, parser, manifest)
 
 
 def make_parser():
@@ -280,24 +364,46 @@ def make_parser():
     )
 
     parser.add_argument(
-        "-l", "--lockdown", type=bool, action="store_true", default=False,
+        "-l", "--lockdown", action="store_true", default=False,
+    )
+    parser.add_argument(
+        "-s", "--strict", action="store_true", default=False,
     )
     parser.add_argument(
-        "-s", "--strict", type=argparse.FileType("r"), default=sys.stdin
+        "--insecure",
+        action="store_true",
+        default=False,
+        help=f"Skip manifest validation (validation currently unsupported)",
     )
     parser.add_argument(
-        "-i", "--input", type=argparse.FileType("r"), default=sys.stdin
+        "--only-validate",
+        action="store_true",
+        default=False,
+        help=f"Exit after validating the manifest (validation currently unsupported)",
     )
     parser.add_argument(
-        "-o", "--output", type=argparse.FileType("w"), default=sys.stdout
+        "--input-action",
+        choices=INPUT_ACTIONS.keys(),
+        default="read_text",
+        help=f"Method for aquiring manifest",
+    )
+    parser.add_argument(
+        "--input-target",
+        help="External callable to run which provides manifest",
+    )
+    parser.add_argument(
+        "--input-args",
+        help="Arguments for input action if externally callable",
     )
-    parser.add_argument("-n", "--name", help="Name of function to replace")
     return parser
 
 
 def main():
     parser = make_parser()
     args = parser.parse_args()
-    args.output.write(
-        replace_function(args.input.read(), args.name, args.func.read()) + "\n"
-    )
+
+    shim(args)
+
+
+if __name__ == "__main__":
+    main()