修复浏览器缓存内部链接复制并冻结门禁候选（09-06-release-gate-integrity）
这个提交包含在：
@@ -0,0 +1,172 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Copy a frozen local browser cache without flattening internal file aliases."""
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import stat
|
||||
import sys
|
||||
|
||||
|
||||
class CacheCopyError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def regular_reader(file):
|
||||
descriptor = os.open(file, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
||||
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
|
||||
os.close(descriptor)
|
||||
raise CacheCopyError('special_file')
|
||||
return os.fdopen(descriptor, 'rb')
|
||||
|
||||
|
||||
def file_digest(file):
|
||||
digest = hashlib.sha256()
|
||||
with regular_reader(file) as stream:
|
||||
for chunk in iter(lambda: stream.read(1024 * 1024), b''):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def snapshot(source):
|
||||
source = Path(source)
|
||||
if source.is_symlink() or not source.is_dir():
|
||||
raise CacheCopyError('source_not_physical_directory')
|
||||
root = source.resolve(strict=True)
|
||||
entries = {}
|
||||
def walk_error(_error):
|
||||
raise CacheCopyError('source_unreadable')
|
||||
# Classify and validate every link before opening any linked file content.
|
||||
for parent, directories, files in os.walk(root, followlinks=False, onerror=walk_error):
|
||||
for name in sorted(directories + files):
|
||||
file = Path(parent) / name
|
||||
mode = file.lstat().st_mode
|
||||
relative = file.relative_to(root).as_posix()
|
||||
if stat.S_ISLNK(mode):
|
||||
target = os.readlink(file)
|
||||
if os.path.isabs(target):
|
||||
raise CacheCopyError('absolute_link')
|
||||
try:
|
||||
resolved = file.resolve(strict=True)
|
||||
except (OSError, RuntimeError):
|
||||
raise CacheCopyError('broken_or_cyclic_link') from None
|
||||
if not resolved.is_relative_to(root):
|
||||
raise CacheCopyError('external_link')
|
||||
if not resolved.is_file():
|
||||
raise CacheCopyError('link_target_not_regular_file')
|
||||
entries[relative] = {'kind': 'link', 'target': target, 'resolved': resolved.relative_to(root).as_posix()}
|
||||
elif stat.S_ISDIR(mode):
|
||||
entries[relative] = {'kind': 'directory', 'mode': stat.S_IMODE(mode)}
|
||||
elif stat.S_ISREG(mode):
|
||||
entries[relative] = {'kind': 'file', 'mode': stat.S_IMODE(mode), 'size': file.stat().st_size}
|
||||
else:
|
||||
raise CacheCopyError('special_file')
|
||||
for name, entry in entries.items():
|
||||
if entry['kind'] == 'file':
|
||||
entry['sha256'] = file_digest(root / name)
|
||||
for entry in entries.values():
|
||||
if entry['kind'] == 'link':
|
||||
target = entries.get(entry['resolved'])
|
||||
if not target or target['kind'] != 'file':
|
||||
raise CacheCopyError('link_target_not_in_manifest')
|
||||
return {'rootMode': stat.S_IMODE(root.stat().st_mode), 'entries': dict(sorted(entries.items()))}
|
||||
|
||||
|
||||
def snapshot_digest(value):
|
||||
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
||||
|
||||
|
||||
def check_paths(source, destination):
|
||||
if source.is_symlink() or not source.is_dir():
|
||||
raise CacheCopyError('source_not_physical_directory')
|
||||
source_root = source.resolve(strict=True)
|
||||
target_root = destination.resolve(strict=False)
|
||||
if target_root.is_relative_to(source_root) or source_root.is_relative_to(target_root):
|
||||
raise CacheCopyError('overlapping_directories')
|
||||
if os.path.lexists(destination):
|
||||
raise CacheCopyError('destination_exists')
|
||||
if not destination.parent.is_dir():
|
||||
raise CacheCopyError('destination_parent_missing')
|
||||
return source_root, target_root
|
||||
|
||||
|
||||
def copy_regular(source, destination):
|
||||
# Exclusive destination and no-follow source: never turn a file alias into
|
||||
# an extra loaded library, or reuse a pre-existing destination inode.
|
||||
with regular_reader(source) as reader, destination.open('xb') as writer:
|
||||
metadata = os.fstat(reader.fileno())
|
||||
shutil.copyfileobj(reader, writer)
|
||||
os.fchmod(writer.fileno(), stat.S_IMODE(metadata.st_mode))
|
||||
os.utime(destination, ns=(metadata.st_atime_ns, metadata.st_mtime_ns), follow_symlinks=False)
|
||||
|
||||
|
||||
def copy_cache(source, destination):
|
||||
source, destination = check_paths(Path(source), Path(destination))
|
||||
before = snapshot(source)
|
||||
destination.mkdir(mode=0o700)
|
||||
directories = [name for name, entry in before['entries'].items() if entry['kind'] == 'directory']
|
||||
for name in sorted(directories, key=lambda value: (value.count('/'), value)):
|
||||
(destination / name).mkdir(mode=0o700)
|
||||
for name, entry in before['entries'].items():
|
||||
if entry['kind'] == 'file':
|
||||
copy_regular(source / name, destination / name)
|
||||
for name, entry in before['entries'].items():
|
||||
if entry['kind'] == 'link':
|
||||
(destination / name).symlink_to(entry['target'])
|
||||
for name in reversed(sorted(directories, key=lambda value: (value.count('/'), value))):
|
||||
(destination / name).chmod(before['entries'][name]['mode'])
|
||||
destination.chmod(before['rootMode'])
|
||||
after = snapshot(source)
|
||||
copied = snapshot(destination)
|
||||
if before != after:
|
||||
raise CacheCopyError('source_changed')
|
||||
if copied != before:
|
||||
raise CacheCopyError('copy_identity_mismatch')
|
||||
regular = [name for name, entry in before['entries'].items() if entry['kind'] == 'file']
|
||||
source_inodes = {(entry.st_dev, entry.st_ino) for entry in ((source / name).stat() for name in regular)}
|
||||
if any((entry.st_dev, entry.st_ino) in source_inodes for entry in ((destination / name).stat() for name in regular)):
|
||||
raise CacheCopyError('source_inode_shared')
|
||||
return {
|
||||
'schemaVersion': 1, 'status': 'PASS', 'scope': 'LOCAL_BROWSER_CACHE_COPY_ONLY',
|
||||
'sourceUnchanged': True, 'sourceInodesIndependent': True, 'bytesModesAndLinksMatch': True,
|
||||
'regularFiles': len(regular), 'internalFileLinks': sum(entry['kind'] == 'link' for entry in before['entries'].values()),
|
||||
'sourceBeforeSha256': snapshot_digest(before), 'sourceAfterSha256': snapshot_digest(after),
|
||||
'destinationSha256': snapshot_digest(copied), 'manifest': before, 'deploymentEligible': False,
|
||||
}
|
||||
|
||||
|
||||
def main(arguments):
|
||||
if len(arguments) != 3:
|
||||
raise CacheCopyError('usage_source_destination_receipt_required')
|
||||
source, destination, receipt = map(Path, arguments)
|
||||
source_root, target_root = check_paths(source, destination)
|
||||
receipt_root = receipt.resolve(strict=False)
|
||||
if receipt_root.is_relative_to(source_root) or receipt_root.is_relative_to(target_root):
|
||||
raise CacheCopyError('receipt_inside_cache')
|
||||
if os.path.lexists(receipt):
|
||||
raise CacheCopyError('receipt_exists')
|
||||
# Preserve failures as new evidence; never erase a partially copied cache.
|
||||
with receipt.open('x', encoding='utf8') as stream:
|
||||
os.fchmod(stream.fileno(), 0o600)
|
||||
try:
|
||||
result = copy_cache(source, destination)
|
||||
except CacheCopyError as error:
|
||||
result = {'status': 'FAIL', 'errorKind': str(error), 'deploymentEligible': False}
|
||||
except OSError:
|
||||
result = {'status': 'FAIL', 'errorKind': 'filesystem_error', 'deploymentEligible': False}
|
||||
json.dump(result, stream, ensure_ascii=False, indent=2)
|
||||
stream.write('\n')
|
||||
print(json.dumps({key: value for key, value in result.items() if key != 'manifest'}))
|
||||
return 0 if result['status'] == 'PASS' else 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
try:
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
except CacheCopyError as error:
|
||||
print(json.dumps({'status': 'FAIL', 'errorKind': str(error), 'deploymentEligible': False}))
|
||||
raise SystemExit(1)
|
||||
except OSError:
|
||||
print(json.dumps({'status': 'FAIL', 'errorKind': 'filesystem_error', 'deploymentEligible': False}))
|
||||
raise SystemExit(1)
|
||||
在新工单中引用
屏蔽一个用户