Python API

Generate a local SBOM bundle for a selected CWL process.

Discover effective Docker requirements along reachable workflow paths.

discover(root, processes)

Return one coverage entry per reachable tool invocation, with inheritance.

Source code in src/cwl2sbom/discovery.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def discover(
    root: dict[str, Any], processes: dict[str, dict[str, Any]]
) -> list[dict[str, Any]]:
    """Return one coverage entry per reachable tool invocation, with inheritance."""
    records: list[dict[str, Any]] = []

    def visit(
        node: dict[str, Any],
        path: str,
        required: dict[str, Any] | None,
        hinted: dict[str, Any] | None,
        ancestors: frozenset[str],
    ) -> None:
        identity = str(node.get("id", path))
        if identity in ancestors:
            raise PluginFailureError(f"Recursive workflow reference: {identity}")
        own_req, own_hint = docker(node, "requirements"), docker(node, "hints")
        required = own_req if own_req is not None else required
        hinted = own_hint if own_hint is not None else hinted
        if node.get("class") == "Workflow":
            for step in node.get("steps", []):
                run = step["run"]
                child = (
                    run
                    if isinstance(run, dict)
                    else (processes.get(run) or processes.get(run.removeprefix("#")))
                )
                if child is None:
                    raise PluginFailureError(f"Unresolved workflow step: {run}")
                step_req, step_hint = (
                    docker(step, "requirements"),
                    docker(step, "hints"),
                )
                visit(
                    child,
                    f"{path}/{step['id']}",
                    step_req if step_req is not None else required,
                    step_hint if step_hint is not None else hinted,
                    ancestors | {identity},
                )
            return
        if node.get("class") == "ExpressionTool":
            records.append(
                {
                    "step": path,
                    "process": identity,
                    "status": "not-applicable",
                    "reason": "ExpressionTool runs in the CWL engine",
                }
            )
            return
        image, reason = image_reference(required if required is not None else hinted)
        records.append(
            {
                "step": path,
                "process": identity,
                "image": image,
                "status": "declared" if image else "uncovered",
                "reason": reason,
                "declaration": "requirement" if required is not None else "hint",
            }
        )

    visit(root, str(root["id"]), None, None, frozenset())
    return records

Invoke Trivy without a shell and retain its original CycloneDX output.