From: Daniel Turull <daniel.turull@ericsson.com>
Date: Tue, 21 Jul 2026 14:48:05 +0000 (+0200)
Subject: scripts/contrib: add helper to report and fix missing SRC_URI ;tag=
X-Git-Url: http://git.ipfire.org/gitweb/?a=commitdiff_plain;h=d9a75fa667a89f9dfd81b8fdcea74c5b9048689b;p=thirdparty%2Fopenembedded%2Fopenembedded-core.git

scripts/contrib: add helper to report and fix missing SRC_URI ;tag=

Add a contrib helper which scans parsed recipes for git/gitsm SRC_URI
entries missing a ;tag= parameter.  By default it reports only and does
not construct the BitBake fetcher.  With --query-remote it queries
upstream tags and, when a well-known tag format resolves to the recipe's
SRCREV, suggests it as a ;tag= candidate.  With --write (which implies
--query-remote) the unambiguous candidate is written back to the recipe.

Tag candidate search uses TAG_FORMATS (v${PV}, ${PV}, ${BPN}-${PV})
followed by UPSTREAM_CHECK_GITTAGREGEX if defined in the recipe.  A
candidate is only accepted when it resolves to the recipe's literal
SRCREV on the remote, so floating revisions (AUTOREV) and non-release
PVs are skipped.

Writes are performed via oe.recipeutils.patch_recipe(), which also
localises the change to the .inc file when SRC_URI is defined there.
Only literal SRC_URI entries (no ${...} in the URL token) are modified;
the entry is edited with bb.fetch2.decodeurl()/encodeurl() so the
;tag= parameter is appended without disturbing the other parameters.

Every modified file is printed, with a reminder to review the result
with git diff and test it with bitbake -c fetch.  With --recipe and
--layer the scan can be restricted to a single recipe or layer, and
--fail-on-missing makes the exit status reflect URLs still missing after
any writes complete.

AI-Generated: Kiro with Claude-opus-4.8
Signed-off-by: Daniel Turull <daniel.turull@ericsson.com>
Signed-off-by: Mathieu Dubois-Briand <mathieu.dubois-briand@bootlin.com>
Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
---

diff --git a/scripts/contrib/check-srcuri-tag.py b/scripts/contrib/check-srcuri-tag.py
new file mode 100755
index 0000000000..33a2b781f7
--- /dev/null
+++ b/scripts/contrib/check-srcuri-tag.py
@@ -0,0 +1,483 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: OpenEmbedded Contributors
+#
+# SPDX-License-Identifier: MIT
+
+# pylint: disable=invalid-name
+
+"""
+Scan parsed recipes for git/gitsm SRC_URI entries missing a ;tag= parameter.
+
+Default mode reports only.  With --query-remote the script queries upstream
+tags and, when a well-known tag format (v${PV}, ${PV}, ${BPN}-${PV} or the
+recipe's UPSTREAM_CHECK_GITTAGREGEX) resolves to the recipe's SRCREV, suggests
+it as a ;tag= candidate.  With --write (which implies --query-remote) the
+unambiguous candidate is written directly into the file that literally
+defines the SRC_URI entry (which may be a .inc file required by the
+recipe).  Only literal SRC_URI entries (no ${...} in the URL token) are
+modified.
+
+Must be run from an initialized build environment (oe-init-build-env).
+
+Usage examples:
+  check-srcuri-tag.py
+  check-srcuri-tag.py --query-remote
+  check-srcuri-tag.py --write
+  check-srcuri-tag.py --recipe busybox --write
+  check-srcuri-tag.py --layer /path/to/meta-foo
+  check-srcuri-tag.py --fail-on-missing
+"""
+
+import argparse
+import logging
+import os
+import re
+import sys
+from dataclasses import dataclass
+
+scripts_path = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
+sys.path.insert(0, os.path.join(scripts_path, 'lib'))
+
+import scriptpath  # pylint: disable=import-error,wrong-import-position
+if not scriptpath.add_bitbake_lib_path():
+    sys.exit("Unable to find bitbake libraries")
+scriptpath.add_oe_lib_path()
+
+import bb.cache        # pylint: disable=wrong-import-position
+import bb.fetch2       # pylint: disable=wrong-import-position
+import bb.tinfoil      # pylint: disable=wrong-import-position
+import oe.recipeutils  # pylint: disable=wrong-import-position
+
+# Well-known tag formats tried in order when --query-remote is active.
+TAG_FORMATS = ["v${PV}", "${PV}", "${BPN}-${PV}"]
+
+# Cache of remote tags keyed by upstream repository identity.
+_lsremote_cache = {}
+
+
+@dataclass(eq=False)
+class UrlResult:
+    """Outcome for a single git URL within a recipe."""
+    entry: str          # unexpanded SRC_URI token
+    base_url: str       # expanded scheme://host/path (no parameters)
+    name: str           # name= parameter, or ""
+    status: str         # tagged|missing|remote-error|parse-error
+    tag: str = ""       # existing ;tag= value (status == tagged)
+    tag_fmt: str = ""   # suggested tag candidate (status == missing)
+    detail: str = ""    # human-readable note
+
+
+def get_literal_srcrev(name, data):
+    """Return (sha, None) or (None, reason) for the named git URL.
+
+    SRCREV candidates are read unexpanded (to avoid triggering AUTOREV
+    resolution) in the same fallback order used by bb.fetch2, accepting only a
+    plain hex SHA (or the SHA embedded in a cached AUTOINC+<sha> value).
+    """
+    pn = data.getVar("PN") or ""
+    attempts = []
+    if name and name != "default":
+        if pn:
+            attempts.append(f"SRCREV_{name}:pn-{pn}")
+        attempts.append(f"SRCREV_{name}")
+    if pn:
+        attempts.append(f"SRCREV:pn-{pn}")
+    attempts.append("SRCREV")
+
+    for key in attempts:
+        val = data.getVar(key, False)
+        if not val or val == "INVALID":
+            continue
+        if val.startswith("AUTOINC+"):
+            val = val[len("AUTOINC+"):]
+        if "AUTOREV" in val or "AUTOINC" in val:
+            return None, "SRCREV is a floating revision"
+        if "${" in val:
+            return None, "SRCREV is not a literal value"
+        val = val.lower()
+        if re.fullmatch(r'[0-9a-f]{40}|[0-9a-f]{64}', val):
+            return val, None
+        return None, "SRCREV is not a literal SHA"
+    return None, "SRCREV is not set"
+
+
+def get_release_pv(data):
+    """Return (pv, None) for a plain release PV, or (None, reason).
+
+    The unexpanded PV is checked first so PV = "${SRCPV}" or "1.0+git" are
+    rejected before any expansion side effects occur.
+    """
+    pv_raw = data.getVar("PV", False) or ""
+    if not pv_raw or "git" in pv_raw or "SRCPV" in pv_raw:
+        return None, f"PV '{pv_raw}' is not a plain release version"
+    return data.getVar("PV") or pv_raw, None
+
+
+def lsremote_tags(ud, data):
+    """Return ({tag_name: frozenset(shas)}, None) or (None, reason).
+
+    Uses the Git fetcher's _lsremote() so mirrors, credentials and
+    BB_NO_NETWORK are honoured.  Both the tag-object and peeled-commit SHAs of
+    annotated tags are collected under the same tag name.
+    """
+    key = (getattr(ud, 'user', ''), getattr(ud, 'proto', ''),
+           getattr(ud, 'host', ''), getattr(ud, 'path', ''))
+    if key in _lsremote_cache:
+        return _lsremote_cache[key]
+
+    try:
+        output = ud.method._lsremote(ud, data, "refs/tags/*")
+    except (bb.fetch2.NetworkAccess, bb.fetch2.FetchError) as exc:
+        result = (None, f"remote error: {exc}")
+        _lsremote_cache[key] = result
+        return result
+
+    tag_map = {}
+    for line in output.splitlines():
+        parts = line.split(None, 1)
+        if len(parts) != 2:
+            continue
+        sha, ref = parts
+        tag = ref.removeprefix("refs/tags/").removesuffix("^{}")
+        tag_map.setdefault(tag, set()).add(sha)
+
+    result = ({k: frozenset(v) for k, v in tag_map.items()}, None)
+    _lsremote_cache[key] = result
+    return result
+
+
+def deduce_tag_format(ud, srcrev, pv, bpn, data):
+    """Return (tag_fmt, None) if a well-known format resolves to srcrev on the
+    remote, else (None, reason).
+
+    Tries TAG_FORMATS first; if none match and the recipe defines
+    UPSTREAM_CHECK_GITTAGREGEX, that regex is used to find the tag whose
+    captured 'pver' group (normalized from '_' to '.') equals PV and whose SHA
+    matches SRCREV.  A unique regex hit is returned as a literal tag name.
+    """
+    tag_map, err = lsremote_tags(ud, data)
+    if tag_map is None:
+        return None, err
+    if not tag_map:
+        return None, "remote has no tags"
+
+    matches = [fmt for fmt in TAG_FORMATS
+               if srcrev in tag_map.get(
+                   fmt.replace("${PV}", pv).replace("${BPN}", bpn), frozenset())]
+    if len(matches) == 1:
+        return matches[0], None
+    if len(matches) > 1:
+        return None, f"ambiguous: multiple formats match ({', '.join(matches)})"
+
+    check_regex = data.getVar("UPSTREAM_CHECK_GITTAGREGEX", False)
+    if check_regex:
+        try:
+            compiled = re.compile(check_regex)
+        except re.error as exc:
+            return None, f"UPSTREAM_CHECK_GITTAGREGEX is invalid: {exc}"
+        regex_matches = []
+        for tag_name, shas in tag_map.items():
+            m = compiled.fullmatch(tag_name)
+            if (m and "pver" in m.groupdict()
+                    and m.group("pver").replace("_", ".") == pv
+                    and srcrev in shas):
+                regex_matches.append(tag_name)
+        if len(regex_matches) == 1:
+            return regex_matches[0], None
+        if len(regex_matches) > 1:
+            return None, ("ambiguous: UPSTREAM_CHECK_GITTAGREGEX matched "
+                          f"multiple tags ({', '.join(sorted(regex_matches))})")
+
+    return None, "no well-known tag format matched SRCREV on remote"
+
+
+def add_candidate(result, exp_url, name, data):
+    """Fill in result.tag_fmt/detail for a missing-tag URL under --query-remote.
+
+    Sets result.status to 'remote-error' if the remote could not be queried.
+    """
+    srcrev, reason = get_literal_srcrev(name or "default", data)
+    if not srcrev:
+        result.detail = f"no suggestion: {reason}"
+        return
+    pv, reason = get_release_pv(data)
+    if not pv:
+        result.detail = f"no suggestion: {reason}"
+        return
+
+    try:
+        ud = bb.fetch2.FetchData(exp_url, data)
+    except bb.fetch2.FetchError as exc:
+        result.status = "remote-error"
+        result.detail = str(exc)
+        return
+
+    fmt, reason = deduce_tag_format(ud, srcrev, pv, data.getVar("BPN") or "", data)
+    if fmt:
+        result.tag_fmt = fmt
+        result.detail = f"candidate: ;tag={fmt}"
+    elif reason.startswith("remote error"):
+        result.status = "remote-error"
+        result.detail = reason
+    else:
+        result.detail = f"no suggestion: {reason}"
+
+
+def check_url(entry, data, args):
+    """Process a single git/gitsm SRC_URI entry.  Returns a UrlResult."""
+    exp = data.expand(entry)
+    try:
+        scheme, host, path, _u, _p, parm = bb.fetch2.decodeurl(exp)
+    except bb.fetch2.MalformedUrl as exc:
+        return UrlResult(entry, exp, "", "parse-error", detail=str(exc))
+
+    base_url = f"{scheme}://{host}{path}"
+    name = parm.get("name", "")
+    # The git fetcher no longer supports comma-separated name= values.
+    if "," in name:
+        return UrlResult(entry, base_url, name, "parse-error",
+                         detail="comma in name= parameter")
+    if parm.get("tag"):
+        return UrlResult(entry, base_url, name, "tagged", tag=parm["tag"])
+
+    result = UrlResult(entry, base_url, name, "missing")
+    if args.query_remote:
+        add_candidate(result, exp, name, data)
+    return result
+
+
+def check_recipe(data, args):
+    """Return a list of UrlResult for every git/gitsm SRC_URI entry."""
+    entries = oe.recipeutils.split_var_value(
+        data.getVar("SRC_URI", False) or "", assignment=False)
+    return [check_url(e, data, args) for e in entries
+            if e.startswith(("git://", "gitsm://"))]
+
+
+def add_tag_to_entry(entry, tag_fmt):
+    """Return entry with ;tag=<tag_fmt> added (entry must be a literal URL)."""
+    decoded = list(bb.fetch2.decodeurl(entry))
+    decoded[5]["tag"] = tag_fmt
+    return bb.fetch2.encodeurl(decoded)
+
+
+def find_source_file(entry, data):
+    """Return (filepath, None) for the one file whose SRC_URI fragment
+    literally contains entry, or (None, reason) if that can't be determined
+    unambiguously.
+
+    entry must contain no variable references (checked by the caller).  Only
+    files recorded in SRC_URI's variable history are considered, and entry
+    must appear as a whole space-separated token in exactly one of them (this
+    also catches the case where the git URL is set in a .inc file that is
+    require'd by the .bb, which is common for git recipes).
+    """
+    matches = []
+    for event in data.varhistory.variable("SRC_URI") or []:
+        fn = event.get("file")
+        detail = event.get("detail")
+        if not fn or not detail:
+            continue
+        if entry in detail.split() and fn not in matches:
+            matches.append(fn)
+
+    if not matches:
+        return None, "URL not found literally in SRC_URI history"
+    if len(matches) > 1:
+        return None, f"URL set in multiple files: {', '.join(matches)}"
+    return matches[0], None
+
+
+def apply_tag(fn, entry, tag_fmt):
+    """Replace entry with entry;tag=<tag_fmt> in fn.  entry must appear as a
+    whole space/quote-delimited token exactly once.
+
+    Returns (True, None) on success, or (False, reason) on error.
+    """
+    try:
+        with open(fn, encoding="utf-8") as f:
+            content = f.read()
+    except OSError as exc:
+        return False, f"read error: {exc}"
+
+    pattern = re.compile(r'(?<![^\s"])%s(?=[\s"\\]|$)' % re.escape(entry))
+    replacement = add_tag_to_entry(entry, tag_fmt)
+    new_content, count = pattern.subn(lambda _m: replacement, content)
+    if count != 1:
+        return False, f"expected 1 occurrence of URL in {fn}, found {count}"
+
+    try:
+        with open(fn, "w", encoding="utf-8") as f:
+            f.write(new_content)
+    except OSError as exc:
+        return False, f"write error: {exc}"
+    return True, None
+
+
+def write_candidates(data, results, layer_path):
+    """Write unambiguous ;tag= candidates directly into their source file(s).
+
+    Only literal SRC_URI entries (no ${...}) are touched.  Returns the list
+    of (UrlResult, filepath) pairs that were written.
+    """
+    written = []
+    for r in results:
+        if r.status != "missing" or not r.tag_fmt:
+            continue
+        if "${" in r.entry:
+            r.detail = f"write skipped: URL contains variable references ({r.entry})"
+            continue
+        fn, reason = find_source_file(r.entry, data)
+        if not fn:
+            r.detail = f"write skipped: {reason}"
+            continue
+        if layer_path and not under_layer(fn, layer_path):
+            r.detail = f"write skipped: {fn} is outside --layer"
+            continue
+        ok, reason = apply_tag(fn, r.entry, r.tag_fmt)
+        if ok:
+            written.append((r, fn))
+        else:
+            r.detail = f"write skipped: {reason}"
+    return written
+
+
+def under_layer(fn, layer_path):
+    """Return True if fn is under layer_path (already realpath'd)."""
+    try:
+        return os.path.commonpath([os.path.realpath(fn), layer_path]) == layer_path
+    except ValueError:
+        return False
+
+
+def main():
+    """Main entry point."""
+    parser = argparse.ArgumentParser(
+        description="Report (and optionally fix) git SRC_URI entries missing ;tag=",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="NOTE: --query-remote (implied by --write) contacts upstream "
+               "git servers.")
+    parser.add_argument("--query-remote", action="store_true",
+                        help="Query upstream remotes to suggest a ;tag= for "
+                             "missing entries (requires network access)")
+    parser.add_argument("--write", action="store_true",
+                        help="Apply unambiguous tag candidates to their source "
+                             "recipe/.inc files; implies --query-remote.  Only "
+                             "literal SRC_URI entries (no ${...}) are modified.")
+    parser.add_argument("--recipe", metavar="PN",
+                        help="Check only this recipe name")
+    parser.add_argument("--layer", metavar="PATH",
+                        help="Check only recipes under this layer directory")
+    parser.add_argument("--fail-on-missing", action="store_true",
+                        help="Exit non-zero if any URLs are still missing ;tag= "
+                             "after this run (remote errors are not counted)")
+    parser.add_argument("--verbose", "-v", action="store_true",
+                        help="Show tagged and parse-error URLs as well")
+    args = parser.parse_args()
+
+    if args.write:
+        args.query_remote = True
+
+    layer_path = None
+    if args.layer:
+        layer_path = os.path.realpath(args.layer)
+        if not os.path.isdir(layer_path):
+            sys.exit(f"error: --layer is not a directory: {args.layer!r}")
+
+    logging.basicConfig(
+        format="%(levelname)s: %(message)s",
+        level=logging.DEBUG if args.verbose else logging.WARNING)
+
+    stats = dict.fromkeys(
+        ("total", "tagged", "missing", "candidate", "written",
+         "remote_error", "parse_error"), 0)
+    modified_files = []
+
+    with bb.tinfoil.Tinfoil(tracking=True) as tinfoil:
+        tinfoil.prepare()
+        pkg_pn = tinfoil.cooker.recipecaches[''].pkg_pn
+        checked_files = set()
+
+        for pn in sorted(pkg_pn):
+            if args.recipe and pn != args.recipe:
+                continue
+
+            for fn in pkg_pn[pn]:
+                realfn, _, _ = bb.cache.virtualfn2realfn(fn)
+                if realfn in checked_files:
+                    continue
+                checked_files.add(realfn)
+                if layer_path and not under_layer(realfn, layer_path):
+                    continue
+
+                data = tinfoil.parse_recipe_file(realfn)
+                results = check_recipe(data, args)
+                if not results:
+                    continue
+
+                if args.write:
+                    written = write_candidates(data, results, layer_path)
+                    for _r, wfn in written:
+                        if wfn not in modified_files:
+                            modified_files.append(wfn)
+                else:
+                    written = []
+                written_results = {r for r, _fn in written}
+
+                basename = os.path.basename(realfn)
+                for r in results:
+                    stats["total"] += 1
+                    label = f"{pn}  {basename}"
+                    if r.name and r.name != "default":
+                        label += f"  name={r.name}"
+                    else:
+                        label += f"  {r.base_url}"
+
+                    if r.status == "tagged":
+                        stats["tagged"] += 1
+                        if args.verbose:
+                            print(f"OK:           {label}  tag={r.tag}")
+                    elif r.status == "missing":
+                        stats["missing"] += 1
+                        if r.tag_fmt:
+                            stats["candidate"] += 1
+                        if r in written_results:
+                            stats["written"] += 1
+                            print(f"WRITTEN:      {label}  ;tag={r.tag_fmt}")
+                        else:
+                            suffix = f"  {r.detail}" if r.detail else ""
+                            print(f"MISSING:      {label}{suffix}")
+                    elif r.status == "remote-error":
+                        stats["remote_error"] += 1
+                        print(f"REMOTE-ERROR: {label}  {r.detail}")
+                    elif r.status == "parse-error":
+                        stats["parse_error"] += 1
+                        if args.verbose:
+                            print(f"PARSE-ERROR:  {label}  {r.detail}")
+
+    print("\n--- Summary ---")
+    print(f"Git URLs checked : {stats['total']}")
+    print(f"  Tagged         : {stats['tagged']}")
+    print(f"  Missing tag    : {stats['missing']}")
+    if args.query_remote:
+        print(f"    candidate found: {stats['candidate']}")
+        print(f"  Remote error   : {stats['remote_error']}")
+    if args.write:
+        print(f"  Written        : {stats['written']}")
+    if stats["parse_error"]:
+        print(f"  Parse error    : {stats['parse_error']}")
+
+    if modified_files:
+        print("\nModified files:")
+        for fn in modified_files:
+            print(f"  {fn}")
+        print("\nReview changes with:  git diff")
+        print("Test affected recipes with:  bitbake -c fetch <recipe>")
+
+    if args.fail_on_missing and stats["missing"] - stats["written"] > 0:
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())