tricu

redistribution.py (9655B)

---

 #!/usr/bin/env python3
 """Collect source/notice evidence for actual ELF and archive inputs. Fail closed."""
 import argparse
 import hashlib
 import json
 from pathlib import Path
 import re
 import shutil
 import tarfile
 
 from portable import dependencies
 
 
 def digest(path):
     with path.open('rb') as stream:
         return hashlib.file_digest(stream, 'sha256').hexdigest()
 
 
 def owner(path, components):
     path = path.resolve(strict=True)
     matches = [row for row in components
                if any(path.is_relative_to(Path(root)) for root in row['roots'])]
     if len(matches) != 1:
         raise ValueError(f'expected one source owner for {path}, found {[r["name"] for r in matches]}')
     return matches[0]
 
 
 def license_name(name):
     if Path(name).suffix.lower() in ('.hs', '.json', '.sh', '.py', '.c', '.h'):
         return False
     return bool(re.match(r'^(copying|licen[cs]es?|copyright|notices?|authors)([._-].*|\d.*)?$',
                          Path(name).name, re.I))
 
 
 def notices(source, component):
     """Preserve named texts and leading C-family copyright/permission blocks.
 
     Header notices cover vendored implementations whose terms live beside code
     (for example cryptonite's SHA-256 implementation). Source-only notices are
     included conservatively and indexed by source component.
     """
     def read_notice(name, stream):
         if license_name(name):
             return stream.read()
         if Path(name).suffix.lower() not in ('.c', '.h', '.cc', '.cpp', '.hpp', '.s'):
             return None
         if component.startswith('glibc-') and Path(name).name not in ('crti.S', 'crtn.S', 'start.S', 'abi-note.c', 'static-reloc.c'):
             return None
         if component.startswith('gcc-') and not re.search(r'/(libgcc|libatomic|include)/|/gcc/crtstuff\.c$', name):
             return None
         if component.startswith('ghc-') and not re.search(r'(^|/)(libraries|rts|includes)/', name):
             return None
         data = stream.read(65536)
         match = re.match(rb'\s*((?:/\*.*?\*/\s*)+)', data, re.S)
         if match and re.search(rb'copyright|permission|redistribution|SPDX-License', match[1], re.I):
             return match[1]
         return None
     if source.is_dir():
         for path in sorted(source.rglob('*')):
             if path.is_file() and not path.is_symlink():
                 with path.open('rb') as stream:
                     data = read_notice(str(path), stream)
                 if data:
                     yield str(path.relative_to(source)), data
     else:
         with tarfile.open(source, 'r:*') as archive:
             for member in archive:
                 if member.isfile():
                     with archive.extractfile(member) as stream:
                         data = read_notice(member.name, stream)
                     if data:
                         yield member.name, data
 
 
 def copy_input(source, target):
     if source.is_dir():
         shutil.copytree(source, target, symlinks=True)
     else:
         shutil.copyfile(source, target)
 
 
 def main():
     p = argparse.ArgumentParser(description=__doc__)
     for name in ('components', 'host', 'link-evidence', 'source', 'nixpkgs', 'flake-utils', 'systems', 'policy', 'output'):
         p.add_argument('--'+name, type=Path, required=True)
     a = p.parse_args()
     components = json.loads(a.components.read_text())
     policy = json.loads(a.policy.read_text())['components']
     link_map = (a.link_evidence/'embedding-link.map').read_text()
     paths = sorted(set(re.findall(r'/nix/store/[^\s()]+?\.(?:a|o)(?=[(\s]|$)', link_map)))
     if not any(path.endswith('.a') for path in paths):
         raise ValueError('archive linkage evidence missing')
     selected = {}
     inputs = []
     for raw in paths:
         path = Path(raw).resolve(strict=True)
         component = owner(path, components)
         selected[component['name']] = component
         inputs.append(dict(kind='link-input', path=str(path), sha256=digest(path), component=component['name']))
     shared = {}
     for name, path in dependencies(a.host/'lib/libtricu-haskell.so').items():
         component = owner(path, components)
         # System glibc is supplied by the destination OS. The link map selects
         # the source materials for glibc startup objects.
         if component['name'].startswith('glibc-'):
             continue
         selected[component['name']] = component
         shared[name] = component['name']
         inputs.append(dict(kind='shared-library', soname=name, path=str(path),
                            sha256=digest(path), component=component['name']))
     unknown = set(selected) - set(policy)
     if unknown:
         raise ValueError(f'license review required for new components: {sorted(unknown)}')
     a.output.mkdir(parents=True, exist_ok=False)
     sources = a.output/'sources'
     sources.mkdir()
     notice_root = a.output/'notices'
     notice_root.mkdir()
     rows = []
     for name, component in sorted(selected.items()):
         if Path(name).name != name:
             raise ValueError('unsafe component identity')
         directory = sources/name
         directory.mkdir()
         upstream = Path(component['source'])
         filename = 'upstream' if upstream.is_dir() else upstream.name
         copy_input(upstream, directory/filename)
         copied_patches = []
         for index, patch in enumerate(component['patches']):
             destination = f'patch-{index:03d}-'+Path(patch).name
             copy_input(Path(patch), directory/destination)
             copied_patches.append(destination)
         if component['cabalRevision']:
             shutil.copyfile(component['cabalRevision'], directory/'revised.cabal')
         (directory/'recipe.json').write_text(json.dumps(component, indent=2, sort_keys=True)+'\n')
         files = []
         unique = {}
         for index, (original, data) in enumerate(notices(upstream, name)):
             # Preserve source path in the manifest; numbered names avoid collisions
             # and prevent arbitrary paths in upstream archive names being written.
             folder = notice_root/name
             folder.mkdir(exist_ok=True)
             checksum = hashlib.sha256(data).hexdigest()
             destination = unique.setdefault(checksum, folder/f'{index:04d}-{Path(original).name}.txt')
             if not destination.exists():
                 destination.write_bytes(data)
             files.append(dict(source_path=original, file=str(destination.relative_to(a.output)),
                               sha256=checksum))
         if not files:
             raise ValueError(f'no license/notice texts found for {name}')
         for required in policy[name]['required_notices']:
             if not any(row['source_path'] == required or row['source_path'].endswith('/'+required) for row in files):
                 raise ValueError(f'missing required notice: {name}: {required}')
         rows.append(dict(name=name, version=component['version'], homepage=component['homepage'],
                          terms=policy[name]['terms'],
                          license_metadata=component['licenseMetadata'], upstream_file=filename,
                          patches=copied_patches, notices=files))
     # Complete authored build recipes, lock, overlays and pinned Nixpkgs recipes
     # accompany upstream sources/patches for inspection and rebuilding.
     copy_input(a.source, sources/'tricu')
     copy_input(a.nixpkgs, sources/'nixpkgs')
     copy_input(a.flake_utils, sources/'flake-utils')
     copy_input(a.systems, sources/'systems')
     shutil.copytree(a.link_evidence, a.output/'link-evidence')
     shutil.copyfile(a.policy, a.output/'redistribution-policy.json')
     shutil.copyfile(a.source/'LICENSE', a.output/'LICENSE')
     shutil.copyfile(a.source/'ext/haskell/REDISTRIBUTION.md', a.output/'REDISTRIBUTION.md')
     report = dict(schema='tricu.redistribution.v1', components=rows, inputs=inputs,
                   shared_owners=shared,
                   host_sha256=digest(a.host/'lib/libtricu-haskell.so'),
                   coverage='Conservative archive/object input coverage, including discarded members; shared runtime inputs')
     (a.output/'components.json').write_text(json.dumps(report, indent=2, sort_keys=True)+'\n')
     index = ['# Third-party notices', '',
              'tricu is licensed under ISC; see LICENSE. The following source distributions',
              'supply linked code or bundled native libraries. Complete upstream notice',
              'texts are retained conservatively, including source-only subcomponents.',
              'The upstream texts govern licensing; Nix metadata provides an index.', '',
              'Matching source materials and build recipes accompany this release.', '']
     for row in rows:
         index += [f'## {row["name"]}', row['homepage'], '', row['terms'], '',
                   f'License/notice texts: `notices/{row["name"]}/`.',
                   f'Source and patch inputs: `sources/{row["name"]}/`.', '']
     (a.output/'THIRD-PARTY-NOTICES.md').write_text('\n'.join(index)+'\n')
     (a.output/'license-components.json').write_text(json.dumps(rows, indent=2, sort_keys=True)+'\n')
     hashes = {str(path.relative_to(a.output)): digest(path)
               for path in sorted(a.output.rglob('*')) if path.is_file() and not path.is_symlink()}
     (a.output/'SHA256.json').write_text(json.dumps(hashes, indent=2, sort_keys=True)+'\n')
     links = {str(path.relative_to(a.output)): str(path.readlink())
              for path in sorted(a.output.rglob('*')) if path.is_symlink()}
     (a.output/'SYMLINKS.json').write_text(json.dumps(links, indent=2, sort_keys=True)+'\n')
 
 
 if __name__ == '__main__':
     main()