tricu

cli.py (5286B)

---

 """Record/check JSON manifests; file bytes are read and hashed only by tricu."""
 import argparse
 from dataclasses import asdict
 import json
 import os
 from pathlib import Path
 import re
 import sys
 import tempfile
 
 from . import FileChecker, FileProblem, DEFAULT_MAX_BYTES
 
 SCHEMA = 'tricu.filecheck.v1'
 MAX_MANIFEST_BYTES = 1024 * 1024
 MAX_FILES = 4096
 
 
 def path_text(text):
     if not isinstance(text, str) or not text or '\0' in text or Path(text).is_absolute():
         raise ValueError('file paths must be nonempty relative Unicode paths without NUL')
     text.encode('utf8')
     return text
 
 
 def unique_object(pairs):
     value = {}
     for key, item in pairs:
         if key in value:
             raise ValueError('duplicate JSON key: '+key)
         value[key] = item
     return value
 
 
 def validate_manifest(value):
     if not isinstance(value, dict) or set(value) != {'schema', 'algorithm', 'files'} or value['schema'] != SCHEMA or value['algorithm'] != 'sha256':
         raise ValueError('unsupported manifest schema')
     if not isinstance(value['files'], list) or not 1 <= len(value['files']) <= MAX_FILES:
         raise ValueError(f'manifest must contain 1..{MAX_FILES} files')
     seen = set()
     for entry in value['files']:
         if not isinstance(entry, dict) or set(entry) != {'path', 'bytes', 'sha256'}:
             raise ValueError('invalid manifest entry')
         name = path_text(entry['path'])
         if name in seen:
             raise ValueError('duplicate manifest path: '+name)
         seen.add(name)
         if type(entry['bytes']) is not int or not 0 <= entry['bytes'] <= 2**63-1:
             raise ValueError('invalid manifest byte count')
         if not isinstance(entry['sha256'], str) or not re.fullmatch('[0-9a-f]{64}', entry['sha256']):
             raise ValueError('invalid manifest SHA-256')
     return value
 
 
 def load_manifest(path):
     with path.open('rb') as stream:
         payload = stream.read(MAX_MANIFEST_BYTES+1)
     if len(payload) > MAX_MANIFEST_BYTES:
         raise ValueError('manifest exceeds 1 MiB')
     return validate_manifest(json.loads(payload.decode('utf8'), object_pairs_hook=unique_object))
 
 
 def publish_manifest(path, manifest):
     payload = (json.dumps(manifest, ensure_ascii=False, indent=2)+'\n').encode('utf8')
     if len(payload) > MAX_MANIFEST_BYTES:
         raise ValueError('manifest exceeds 1 MiB')
     # Complete file publication with no replacement. Errors never truncate an
     # existing manifest. Hard-link support on this filesystem is required.
     with tempfile.NamedTemporaryFile(dir=path.parent, prefix='.filecheck-') as stage:
         stage.write(payload)
         stage.flush()
         os.fsync(stage.fileno())
         os.link(stage.name, path)
 
 
 def inspect_files(checker, parent, entries, limit, verify):
     rows = []
     for entry in entries:
         name = entry['path']
         try:
             actual = asdict(checker.hash_file(parent/name, limit))
             if verify:
                 expected = {key: entry[key] for key in ('bytes', 'sha256')}
                 rows.append(dict(path=name, status='ok' if actual == expected else 'changed', expected=expected, actual=actual))
             else:
                 rows.append(dict(path=name, status='ok', **actual))
         except FileProblem as error:
             rows.append(dict(path=name, status='error', error=asdict(error.problem)))
     return rows
 
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     subparsers = parser.add_subparsers(dest='command', required=True)
     for command in ('record', 'verify'):
         sub = subparsers.add_parser(command)
         sub.add_argument('manifest', type=Path)
         sub.add_argument('--max-bytes', type=int, default=DEFAULT_MAX_BYTES)
         if command == 'record':
             sub.add_argument('paths', nargs='+', help='paths relative to the manifest directory')
     args = parser.parse_args()
     try:
         if not 0 <= args.max_bytes <= 2**63-1:
             raise ValueError('max-bytes must be in 0..2**63-1')
         manifest = args.manifest.absolute()
         if args.command == 'record':
             names = [path_text(name) for name in args.paths]
             if len(names) > MAX_FILES or len(names) != len(set(names)):
                 raise ValueError(f'provide 1..{MAX_FILES} distinct paths')
             entries = [dict(path=name) for name in names]
         else:
             entries = load_manifest(manifest)['files']
         with FileChecker() as checker:
             rows = inspect_files(checker, manifest.parent, entries, args.max_bytes, args.command == 'verify')
         code = 2 if any(row['status'] == 'error' for row in rows) else 1 if any(row['status'] == 'changed' for row in rows) else 0
         if args.command == 'record' and code == 0:
             output = dict(schema=SCHEMA, algorithm='sha256', files=[{key: row[key] for key in ('path', 'bytes', 'sha256')} for row in rows])
             publish_manifest(manifest, validate_manifest(output))
         print(json.dumps(dict(ok=code == 0, files=rows), ensure_ascii=False))
         return code
     except (OSError, ValueError, RuntimeError) as error:
         print(json.dumps(dict(ok=False, error=str(error)), ensure_ascii=True))
         return 2
 
 
 if __name__ == '__main__':
     sys.exit(main())