def do_create_spdx(d):
    from datetime import datetime, timezone
    import oe.sbom
    import oe.spdx
    import uuid
    from pathlib import Path
    from contextlib import contextmanager
    import oe.cve_check

    @contextmanager
    def optional_tarfile(name, guard, mode="w"):
        import tarfile
        import bb.compress.zstd

        num_threads = int(d.getVar("BB_NUMBER_THREADS"))

        if guard:
            name.parent.mkdir(parents=True, exist_ok=True)
            with bb.compress.zstd.open(name, mode=mode + "b", num_threads=num_threads) as f:
                with tarfile.open(fileobj=f, mode=mode + "|") as tf:
                    yield tf
        else:
            yield None


    deploy_dir_spdx = Path(d.getVar("DEPLOY_DIR_SPDX"))
    spdx_workdir = Path(d.getVar("SPDXWORK"))
    include_sources = d.getVar("SPDX_INCLUDE_SOURCES") == "1"
    archive_sources = d.getVar("SPDX_ARCHIVE_SOURCES") == "1"
    archive_packaged = d.getVar("SPDX_ARCHIVE_PACKAGED") == "1"
    pkg_arch = d.getVar("SSTATE_PKGARCH")

    creation_time = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    doc = oe.spdx.SPDXDocument()

    doc.name = "recipe-" + d.getVar("PN")
    doc.documentNamespace = get_doc_namespace(d, doc)
    doc.creationInfo.created = creation_time
    doc.creationInfo.comment = "This document was created by analyzing recipe files during the build."
    doc.creationInfo.licenseListVersion = d.getVar("SPDX_LICENSE_DATA")["licenseListVersion"]
    doc.creationInfo.creators.append("Tool: OpenEmbedded Core create-spdx.bbclass")
    doc.creationInfo.creators.append("Organization: %s" % d.getVar("SPDX_ORG"))
    doc.creationInfo.creators.append("Person: N/A ()")

    recipe = oe.spdx.SPDXPackage()
    recipe.name = d.getVar("PN")
    recipe.versionInfo = d.getVar("PV")
    recipe.SPDXID = oe.sbom.get_recipe_spdxid(d)
    recipe.supplier = d.getVar("SPDX_SUPPLIER")
    if bb.data.inherits_class("native", d) or bb.data.inherits_class("cross", d):
        recipe.annotations.append(create_annotation(d, "isNative"))

    homepage = d.getVar("HOMEPAGE")
    if homepage:
        recipe.homepage = homepage

    license = d.getVar("LICENSE")
    if license:
        recipe.licenseDeclared = convert_license_to_spdx(license, doc, d)

    summary = d.getVar("SUMMARY")
    if summary:
        recipe.summary = summary

    description = d.getVar("DESCRIPTION")
    if description:
        recipe.description = description

    if d.getVar("SPDX_CUSTOM_ANNOTATION_VARS"):
        for var in d.getVar('SPDX_CUSTOM_ANNOTATION_VARS').split():
            recipe.annotations.append(create_annotation(d, var + "=" + d.getVar(var)))

    # Some CVEs may be patched during the build process without incrementing the version number,
    # so querying for CVEs based on the CPE id can lead to false positives. To account for this,
    # save the CVEs fixed by patches to source information field in the SPDX.
    patched_cves = oe.cve_check.get_patched_cves(d)
    patched_cves = list(patched_cves)
    patched_cves = ' '.join(patched_cves)
    if patched_cves:
        recipe.sourceInfo = "CVEs fixed: " + patched_cves

    cpe_ids = oe.cve_check.get_cpe_ids(d.getVar("CVE_PRODUCT"), d.getVar("CVE_VERSION"))
    if cpe_ids:
        for cpe_id in cpe_ids:
            cpe = oe.spdx.SPDXExternalReference()
            cpe.referenceCategory = "SECURITY"
            cpe.referenceType = "http://spdx.org/rdf/references/cpe23Type"
            cpe.referenceLocator = cpe_id
            recipe.externalRefs.append(cpe)

    doc.packages.append(recipe)
    doc.add_relationship(doc, "DESCRIBES", recipe)

    add_download_packages(d, doc, recipe)

    if process_sources(d) and include_sources:
        recipe_archive = deploy_dir_spdx / "recipes" / (doc.name + ".tar.zst")
        with optional_tarfile(recipe_archive, archive_sources) as archive:
            spdx_get_src(d)

            add_package_files(
                d,
                doc,
                recipe,
                spdx_workdir,
                lambda file_counter: "SPDXRef-SourceFile-%s-%d" % (d.getVar("PN"), file_counter),
                lambda filepath: ["SOURCE"],
                ignore_dirs=[".git"],
                ignore_top_level_dirs=["temp"],
                archive=archive,
            )

            if archive is not None:
                recipe.packageFileName = str(recipe_archive.name)

    dep_recipes = collect_dep_recipes(d, doc, recipe)

    doc_sha1 = oe.sbom.write_doc(d, doc, pkg_arch, "recipes", indent=get_json_indent(d))
    dep_recipes.append(oe.sbom.DepRecipe(doc, doc_sha1, recipe))

    recipe_ref = oe.spdx.SPDXExternalDocumentRef()
    recipe_ref.externalDocumentId = "DocumentRef-recipe-" + recipe.name
    recipe_ref.spdxDocument = doc.documentNamespace
    recipe_ref.checksum.algorithm = "SHA1"
    recipe_ref.checksum.checksumValue = doc_sha1

    sources = collect_dep_sources(d, dep_recipes)
    found_licenses = {license.name:recipe_ref.externalDocumentId + ":" + license.licenseId for license in doc.hasExtractedLicensingInfos}

    if not recipe_spdx_is_native(d, recipe):
        bb.build.exec_func("read_subpackage_metadata", d)

        pkgdest = Path(d.getVar("PKGDEST"))
        for package in d.getVar("PACKAGES").split():
            if not oe.packagedata.packaged(package, d):
                continue

            package_doc = oe.spdx.SPDXDocument()
            pkg_name = d.getVar("PKG:%s" % package) or package
            package_doc.name = pkg_name
            package_doc.documentNamespace = get_doc_namespace(d, package_doc)
            package_doc.creationInfo.created = creation_time
            package_doc.creationInfo.comment = "This document was created by analyzing packages created during the build."
            package_doc.creationInfo.licenseListVersion = d.getVar("SPDX_LICENSE_DATA")["licenseListVersion"]
            package_doc.creationInfo.creators.append("Tool: OpenEmbedded Core create-spdx.bbclass")
            package_doc.creationInfo.creators.append("Organization: %s" % d.getVar("SPDX_ORG"))
            package_doc.creationInfo.creators.append("Person: N/A ()")
            package_doc.externalDocumentRefs.append(recipe_ref)

            package_license = d.getVar("LICENSE:%s" % package) or d.getVar("LICENSE")

            spdx_package = oe.spdx.SPDXPackage()

            spdx_package.SPDXID = oe.sbom.get_package_spdxid(pkg_name)
            spdx_package.name = pkg_name
            spdx_package.versionInfo = d.getVar("PV")
            spdx_package.licenseDeclared = convert_license_to_spdx(package_license, package_doc, d, found_licenses)
            spdx_package.supplier = d.getVar("SPDX_SUPPLIER")

            package_doc.packages.append(spdx_package)

            package_doc.add_relationship(spdx_package, "GENERATED_FROM", "%s:%s" % (recipe_ref.externalDocumentId, recipe.SPDXID))
            package_doc.add_relationship(package_doc, "DESCRIBES", spdx_package)

            package_archive = deploy_dir_spdx / "packages" / (package_doc.name + ".tar.zst")
            with optional_tarfile(package_archive, archive_packaged) as archive:
                package_files = add_package_files(
                    d,
                    package_doc,
                    spdx_package,
                    pkgdest / package,
                    lambda file_counter: oe.sbom.get_packaged_file_spdxid(pkg_name, file_counter),
                    lambda filepath: ["BINARY"],
                    ignore_top_level_dirs=['CONTROL', 'DEBIAN'],
                    archive=archive,
                )

                if archive is not None:
                    spdx_package.packageFileName = str(package_archive.name)

            add_package_sources_from_debug(d, package_doc, spdx_package, package, package_files, sources)

            oe.sbom.write_doc(d, package_doc, pkg_arch, "packages", indent=get_json_indent(d))

do_create_spdx(d)

def recipe_spdx_is_native(d, recipe):
    return any(a.annotationType == "OTHER" and
      a.annotator == "Tool: %s - %s" % (d.getVar("SPDX_TOOL_NAME"), d.getVar("SPDX_TOOL_VERSION")) and
      a.comment == "isNative" for a in recipe.annotations)

def add_package_sources_from_debug(d, package_doc, spdx_package, package, package_files, sources):
    from pathlib import Path
    import hashlib
    import oe.packagedata
    import oe.spdx

    debug_search_paths = [
        Path(d.getVar('PKGD')),
        Path(d.getVar('STAGING_DIR_TARGET')),
        Path(d.getVar('STAGING_DIR_NATIVE')),
        Path(d.getVar('STAGING_KERNEL_DIR')),
    ]

    pkg_data = oe.packagedata.read_subpkgdata_extended(package, d)

    if pkg_data is None:
        return

    for file_path, file_data in pkg_data["files_info"].items():
        if not "debugsrc" in file_data:
            continue

        for pkg_file in package_files:
            if file_path.lstrip("/") == pkg_file.fileName.lstrip("/"):
                break
        else:
            bb.fatal("No package file found for %s in %s; SPDX found: %s" % (str(file_path), package,
                " ".join(p.fileName for p in package_files)))
            continue

        for debugsrc in file_data["debugsrc"]:
            ref_id = "NOASSERTION"
            for search in debug_search_paths:
                if debugsrc.startswith("/usr/src/kernel"):
                    debugsrc_path = search / debugsrc.replace('/usr/src/kernel/', '')
                else:
                    debugsrc_path = search / debugsrc.lstrip("/")
                # We can only hash files below, skip directories, links, etc.
                if not os.path.isfile(debugsrc_path):
                    continue

                file_sha256 = bb.utils.sha256_file(debugsrc_path)

                if file_sha256 in sources:
                    source_file = sources[file_sha256]

                    doc_ref = package_doc.find_external_document_ref(source_file.doc.documentNamespace)
                    if doc_ref is None:
                        doc_ref = oe.spdx.SPDXExternalDocumentRef()
                        doc_ref.externalDocumentId = "DocumentRef-dependency-" + source_file.doc.name
                        doc_ref.spdxDocument = source_file.doc.documentNamespace
                        doc_ref.checksum.algorithm = "SHA1"
                        doc_ref.checksum.checksumValue = source_file.doc_sha1
                        package_doc.externalDocumentRefs.append(doc_ref)

                    ref_id = "%s:%s" % (doc_ref.externalDocumentId, source_file.file.SPDXID)
                else:
                    bb.debug(1, "Debug source %s with SHA256 %s not found in any dependency" % (str(debugsrc_path), file_sha256))
                break
            else:
                bb.debug(1, "Debug source %s not found" % debugsrc)

            package_doc.add_relationship(pkg_file, "GENERATED_FROM", ref_id, comment=debugsrc)

def spdx_get_src(d):
    """
    save patched source of the recipe in SPDX_WORKDIR.
    """
    import shutil
    spdx_workdir = d.getVar('SPDXWORK')
    spdx_sysroot_native = d.getVar('STAGING_DIR_NATIVE')
    pn = d.getVar('PN')

    workdir = d.getVar("WORKDIR")

    try:
        # The kernel class functions require it to be on work-shared, so we dont change WORKDIR
        if not is_work_shared_spdx(d):
            # Change the WORKDIR to make do_unpack do_patch run in another dir.
            d.setVar('WORKDIR', spdx_workdir)
            # Restore the original path to recipe's native sysroot (it's relative to WORKDIR).
            d.setVar('STAGING_DIR_NATIVE', spdx_sysroot_native)

            # The changed 'WORKDIR' also caused 'B' changed, create dir 'B' for the
            # possibly requiring of the following tasks (such as some recipes's
            # do_patch required 'B' existed).
            bb.utils.mkdirhier(d.getVar('B'))

            bb.build.exec_func('do_unpack', d)
        # Copy source of kernel to spdx_workdir
        if is_work_shared_spdx(d):
            share_src = d.getVar('WORKDIR')
            d.setVar('WORKDIR', spdx_workdir)
            d.setVar('STAGING_DIR_NATIVE', spdx_sysroot_native)
            src_dir = spdx_workdir + "/" + d.getVar('BP')
            bb.utils.mkdirhier(src_dir)
            if bb.data.inherits_class('kernel',d):
                share_src = d.getVar('STAGING_KERNEL_DIR')
            cmd_copy_share = "cp -rf " + share_src + "/* " + src_dir + "/"
            cmd_copy_shared_res = os.popen(cmd_copy_share).read()
            bb.note("cmd_copy_shared_result = " + cmd_copy_shared_res)

            git_path = src_dir + "/.git"
            if os.path.exists(git_path):
                shutils.rmtree(git_path)

        # Make sure gcc and kernel sources are patched only once
        if not (d.getVar('SRC_URI') == "" or is_work_shared_spdx(d)):
            bb.build.exec_func('do_patch', d)

        # Some userland has no source.
        if not os.path.exists( spdx_workdir ):
            bb.utils.mkdirhier(spdx_workdir)
    finally:
        d.setVar("WORKDIR", workdir)

def convert_license_to_spdx(lic, document, d, existing={}):
    from pathlib import Path
    import oe.spdx

    license_data = d.getVar("SPDX_LICENSE_DATA")
    extracted = {}

    def add_extracted_license(ident, name):
        nonlocal document

        if name in extracted:
            return

        extracted_info = oe.spdx.SPDXExtractedLicensingInfo()
        extracted_info.name = name
        extracted_info.licenseId = ident
        extracted_info.extractedText = None

        if name == "PD":
            # Special-case this.
            extracted_info.extractedText = "Software released to the public domain"
        else:
            # Seach for the license in COMMON_LICENSE_DIR and LICENSE_PATH
            for directory in [d.getVar('COMMON_LICENSE_DIR')] + (d.getVar('LICENSE_PATH') or '').split():
                try:
                    with (Path(directory) / name).open(errors="replace") as f:
                        extracted_info.extractedText = f.read()
                        break
                except FileNotFoundError:
                    pass
            if extracted_info.extractedText is None:
                # If it's not SPDX or PD, then NO_GENERIC_LICENSE must be set
                filename = d.getVarFlag('NO_GENERIC_LICENSE', name)
                if filename:
                    filename = d.expand("${S}/" + filename)
                    with open(filename, errors="replace") as f:
                        extracted_info.extractedText = f.read()
                else:
                    bb.fatal("Cannot find any text for license %s" % name)

        extracted[name] = extracted_info
        document.hasExtractedLicensingInfos.append(extracted_info)

    def convert(l):
        if l == "(" or l == ")":
            return l

        if l == "&":
            return "AND"

        if l == "|":
            return "OR"

        if l == "CLOSED":
            return "NONE"

        spdx_license = d.getVarFlag("SPDXLICENSEMAP", l) or l
        if spdx_license in license_data["licenses"]:
            return spdx_license

        try:
            spdx_license = existing[l]
        except KeyError:
            spdx_license = "LicenseRef-" + l
            add_extracted_license(spdx_license, l)

        return spdx_license

    lic_split = lic.replace("(", " ( ").replace(")", " ) ").replace("|", " | ").replace("&", " & ").split()

    return ' '.join(convert(l) for l in lic_split)

def add_package_files(d, doc, spdx_pkg, topdir, get_spdxid, get_types, *, archive=None, ignore_dirs=[], ignore_top_level_dirs=[]):
    from pathlib import Path
    import oe.spdx
    import hashlib

    source_date_epoch = d.getVar("SOURCE_DATE_EPOCH")
    if source_date_epoch:
        source_date_epoch = int(source_date_epoch)

    sha1s = []
    spdx_files = []

    file_counter = 1

    check_compiled_sources = d.getVar("SPDX_INCLUDE_COMPILED_SOURCES") == "1"
    if check_compiled_sources:
        compiled_sources, types = oe.spdx.get_compiled_sources(d)
        bb.debug(1, f"Total compiled files: {len(compiled_sources)}")
    for subdir, dirs, files in os.walk(topdir):
        dirs[:] = [d for d in dirs if d not in ignore_dirs]
        if subdir == str(topdir):
            dirs[:] = [d for d in dirs if d not in ignore_top_level_dirs]

        for file in files:
            filepath = Path(subdir) / file
            filename = str(filepath.relative_to(topdir))

            if not filepath.is_symlink() and filepath.is_file():
                # Check if file is compiled
                if check_compiled_sources:
                     if not oe.spdx.is_compiled_source(filename, compiled_sources, types):
                          continue
                spdx_file = oe.spdx.SPDXFile()
                spdx_file.SPDXID = get_spdxid(file_counter)
                for t in get_types(filepath):
                    spdx_file.fileTypes.append(t)
                spdx_file.fileName = filename

                if archive is not None:
                    with filepath.open("rb") as f:
                        info = archive.gettarinfo(fileobj=f)
                        info.name = filename
                        info.uid = 0
                        info.gid = 0
                        info.uname = "root"
                        info.gname = "root"

                        if source_date_epoch is not None and info.mtime > source_date_epoch:
                            info.mtime = source_date_epoch

                        archive.addfile(info, f)

                sha1 = bb.utils.sha1_file(filepath)
                sha1s.append(sha1)
                spdx_file.checksums.append(oe.spdx.SPDXChecksum(
                        algorithm="SHA1",
                        checksumValue=sha1,
                    ))
                spdx_file.checksums.append(oe.spdx.SPDXChecksum(
                        algorithm="SHA256",
                        checksumValue=bb.utils.sha256_file(filepath),
                    ))

                if "SOURCE" in spdx_file.fileTypes:
                    extracted_lics = extract_licenses(filepath)
                    if extracted_lics:
                        spdx_file.licenseInfoInFiles = extracted_lics

                doc.files.append(spdx_file)
                doc.add_relationship(spdx_pkg, "CONTAINS", spdx_file)
                spdx_pkg.hasFiles.append(spdx_file.SPDXID)

                spdx_files.append(spdx_file)

                file_counter += 1

    sha1s.sort()
    verifier = hashlib.sha1()
    for v in sha1s:
        verifier.update(v.encode("utf-8"))
    spdx_pkg.packageVerificationCode.packageVerificationCodeValue = verifier.hexdigest()

    return spdx_files

def create_annotation(d, comment):
    from datetime import datetime, timezone

    creation_time = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    annotation = oe.spdx.SPDXAnnotation()
    annotation.annotationDate = creation_time
    annotation.annotationType = "OTHER"
    annotation.annotator = "Tool: %s - %s" % (d.getVar("SPDX_TOOL_NAME"), d.getVar("SPDX_TOOL_VERSION"))
    annotation.comment = comment
    return annotation

def collect_dep_recipes(d, doc, spdx_recipe):
    import json
    from pathlib import Path
    import oe.sbom
    import oe.spdx

    deploy_dir_spdx = Path(d.getVar("DEPLOY_DIR_SPDX"))
    spdx_deps_file = Path(d.getVar("SPDXDEPS"))
    package_archs = d.getVar("SPDX_MULTILIB_SSTATE_ARCHS").split()
    package_archs.reverse()

    dep_recipes = []

    with spdx_deps_file.open("r") as f:
        deps = json.load(f)

    for dep_pn, dep_hashfn in deps:
        dep_recipe_path = oe.sbom.doc_find_by_hashfn(deploy_dir_spdx, package_archs, "recipe-" + dep_pn, dep_hashfn)
        if not dep_recipe_path:
            bb.fatal("Cannot find any SPDX file for recipe %s, %s" % (dep_pn, dep_hashfn))

        spdx_dep_doc, spdx_dep_sha1 = oe.sbom.read_doc(dep_recipe_path)

        for pkg in spdx_dep_doc.packages:
            if pkg.name == dep_pn:
                spdx_dep_recipe = pkg
                break
        else:
            continue

        dep_recipes.append(oe.sbom.DepRecipe(spdx_dep_doc, spdx_dep_sha1, spdx_dep_recipe))

        dep_recipe_ref = oe.spdx.SPDXExternalDocumentRef()
        dep_recipe_ref.externalDocumentId = "DocumentRef-dependency-" + spdx_dep_doc.name
        dep_recipe_ref.spdxDocument = spdx_dep_doc.documentNamespace
        dep_recipe_ref.checksum.algorithm = "SHA1"
        dep_recipe_ref.checksum.checksumValue = spdx_dep_sha1

        doc.externalDocumentRefs.append(dep_recipe_ref)

        doc.add_relationship(
            "%s:%s" % (dep_recipe_ref.externalDocumentId, spdx_dep_recipe.SPDXID),
            "BUILD_DEPENDENCY_OF",
            spdx_recipe
        )

    return dep_recipes

def collect_dep_sources(d, dep_recipes):
    import oe.sbom

    sources = {}
    for dep in dep_recipes:
        # Don't collect sources from native recipes as they
        # match non-native sources also.
        if recipe_spdx_is_native(d, dep.recipe):
            continue
        recipe_files = set(dep.recipe.hasFiles)

        for spdx_file in dep.doc.files:
            if spdx_file.SPDXID not in recipe_files:
                continue

            if "SOURCE" in spdx_file.fileTypes:
                for checksum in spdx_file.checksums:
                    if checksum.algorithm == "SHA256":
                        sources[checksum.checksumValue] = oe.sbom.DepSource(dep.doc, dep.doc_sha1, dep.recipe, spdx_file)
                        break

    return sources

def process_sources(d):
    pn = d.getVar('PN')
    assume_provided = (d.getVar("ASSUME_PROVIDED") or "").split()
    if pn in assume_provided:
        for p in d.getVar("PROVIDES").split():
            if p != pn:
                pn = p
                break

    # glibc-locale: do_fetch, do_unpack and do_patch tasks have been deleted,
    # so avoid archiving source here.
    if pn.startswith('glibc-locale'):
        return False
    if d.getVar('PN') == "libtool-cross":
        return False
    if d.getVar('PN') == "libgcc-initial":
        return False
    if d.getVar('PN') == "shadow-sysroot":
        return False

    # We just archive gcc-source for all the gcc related recipes
    if d.getVar('BPN') in ['gcc', 'libgcc']:
        bb.debug(1, 'spdx: There is bug in scan of %s is, do nothing' % pn)
        return False

    return True

def get_json_indent(d):
    if d.getVar("SPDX_PRETTY") == "1":
        return 2
    return None

def get_doc_namespace(d, doc):
    import uuid
    namespace_uuid = uuid.uuid5(uuid.NAMESPACE_DNS, d.getVar("SPDX_UUID_NAMESPACE"))
    return "%s/%s-%s" % (d.getVar("SPDX_NAMESPACE_PREFIX"), doc.name, str(uuid.uuid5(namespace_uuid, doc.name)))

def read_subpackage_metadata(d):
    import oe.packagedata

    vars = {
        "PN" : d.getVar('PN'),
        "PE" : d.getVar('PE'),
        "PV" : d.getVar('PV'),
        "PR" : d.getVar('PR'),
    }

    data = oe.packagedata.read_pkgdata(vars["PN"], d)

    for key in data.keys():
        d.setVar(key, data[key])

    for pkg in d.getVar('PACKAGES').split():
        sdata = oe.packagedata.read_subpkgdata(pkg, d)
        for key in sdata.keys():
            if key in vars:
                if sdata[key] != vars[key]:
                    if key == "PN":
                        bb.fatal("Recipe %s is trying to create package %s which was already written by recipe %s. This will cause corruption, please resolve this and only provide the package from one recipe or the other or only build one of the recipes." % (vars[key], pkg, sdata[key]))
                    bb.fatal("Recipe %s is trying to change %s from '%s' to '%s'. This will cause do_package_write_* failures since the incorrect data will be used and they will be unable to find the right workdir." % (vars["PN"], key, vars[key], sdata[key]))
                continue
            #
            # If we set unsuffixed variables here there is a chance they could clobber override versions
            # of that variable, e.g. DESCRIPTION could clobber DESCRIPTION:<pkgname>
            # We therefore don't clobber for the unsuffixed variable versions
            #
            if key.endswith(":" + pkg):
                d.setVar(key, sdata[key])
            else:
                d.setVar(key, sdata[key], parsing=True)

def add_download_packages(d, doc, recipe):
    import os.path
    from bb.fetch2 import decodeurl, CHECKSUM_LIST
    import bb.process
    import oe.spdx
    import oe.sbom

    for download_idx, src_uri in enumerate(d.getVar('SRC_URI').split()):
        f = bb.fetch2.FetchData(src_uri, d)

        for name in f.names:
            package = oe.spdx.SPDXPackage()
            package.name = "%s-source-%d" % (d.getVar("PN"), download_idx + 1)
            package.SPDXID = oe.sbom.get_download_spdxid(d, download_idx + 1)

            if f.type == "file":
                continue

            uri = f.type
            proto = getattr(f, "proto", None)
            if proto is not None:
                uri = uri + "+" + proto
            uri = uri + "://" + f.host + f.path

            if f.method.supports_srcrev():
                uri = uri + "@" + f.revisions[name]

            if f.method.supports_checksum(f):
                for checksum_id in CHECKSUM_LIST:
                    if checksum_id.upper() not in oe.spdx.SPDXPackage.ALLOWED_CHECKSUMS:
                        continue

                    expected_checksum = getattr(f, "%s_expected" % checksum_id)
                    if expected_checksum is None:
                        continue

                    c = oe.spdx.SPDXChecksum()
                    c.algorithm = checksum_id.upper()
                    c.checksumValue = expected_checksum
                    package.checksums.append(c)

            package.downloadLocation = uri
            doc.packages.append(package)
            doc.add_relationship(doc, "DESCRIBES", package)
            # In the future, we might be able to do more fancy dependencies,
            # but this should be sufficient for now
            doc.add_relationship(package, "BUILD_DEPENDENCY_OF", recipe)

def extract_licenses(filename):
    import re

    lic_regex = re.compile(rb'^\W*SPDX-License-Identifier:\s*([ \w\d.()+-]+?)(?:\s+\W*)?$', re.MULTILINE)

    try:
        with open(filename, 'rb') as f:
            size = min(15000, os.stat(filename).st_size)
            txt = f.read(size)
            licenses = re.findall(lic_regex, txt)
            if licenses:
                ascii_licenses = [lic.decode('ascii') for lic in licenses]
                return ascii_licenses
    except Exception as e:
        bb.warn(f"Exception reading {filename}: {e}")
    return None

def is_work_shared_spdx(d):
    return bb.data.inherits_class('kernel', d) or ('work-shared' in d.getVar('WORKDIR'))

def do_unpack(d):
    # Export function set
    bb.build.exec_func('base_do_unpack', d)

def do_patch(d):
    # Export function set
    bb.build.exec_func('patch_do_patch', d)

def base_do_unpack(d):
    src_uri = (d.getVar('SRC_URI') or "").split()
    if not src_uri:
        return

    try:
        fetcher = bb.fetch2.Fetch(src_uri, d)
        fetcher.unpack(d.getVar('WORKDIR'))
    except bb.fetch2.BBFetchException as e:
        bb.fatal("Bitbake Fetcher Error: " + repr(e))

def patch_do_patch(d):
    import oe.patch

    patchsetmap = {
        "patch": oe.patch.PatchTree,
        "quilt": oe.patch.QuiltTree,
        "git": oe.patch.GitApplyTree,
    }

    cls = patchsetmap[d.getVar('PATCHTOOL') or 'quilt']

    resolvermap = {
        "noop": oe.patch.NOOPResolver,
        "user": oe.patch.UserResolver,
    }

    rcls = resolvermap[d.getVar('PATCHRESOLVE') or 'user']

    classes = {}

    s = d.getVar('S')

    os.putenv('PATH', d.getVar('PATH'))

    # We must use one TMPDIR per process so that the "patch" processes
    # don't generate the same temp file name.

    import tempfile
    process_tmpdir = tempfile.mkdtemp()
    os.environ['TMPDIR'] = process_tmpdir

    for patch in src_patches(d):
        _, _, local, _, _, parm = bb.fetch.decodeurl(patch)

        if "patchdir" in parm:
            patchdir = parm["patchdir"]
            if not os.path.isabs(patchdir):
                patchdir = os.path.join(s, patchdir)
            if not os.path.isdir(patchdir):
                bb.fatal("Target directory '%s' not found, patchdir '%s' is incorrect in patch file '%s'" %
                    (patchdir, parm["patchdir"], parm['patchname']))
        else:
            patchdir = s

        if not patchdir in classes:
            patchset = cls(patchdir, d)
            resolver = rcls(patchset, oe_terminal)
            classes[patchdir] = (patchset, resolver)
            patchset.Clean()
        else:
            patchset, resolver = classes[patchdir]

        bb.note("Applying patch '%s' (%s)" % (parm['patchname'], oe.path.format_display(local, d)))
        try:
            patchset.Import({"file":local, "strippath": parm['striplevel']}, True)
        except Exception as exc:
            bb.utils.remove(process_tmpdir, True)
            bb.fatal("Importing patch '%s' with striplevel '%s'\n%s" % (parm['patchname'], parm['striplevel'], repr(exc).replace("\\n", "\n")))
        try:
            resolver.Resolve()
        except bb.BBHandledException as e:
            bb.utils.remove(process_tmpdir, True)
            bb.fatal("Applying patch '%s' on target directory '%s'\n%s" % (parm['patchname'], patchdir, repr(e).replace("\\n", "\n")))

    bb.utils.remove(process_tmpdir, True)
    del os.environ['TMPDIR']

def src_patches(d, all=False, expand=True):
    import oe.patch
    return oe.patch.src_patches(d, all, expand)