sashatrask
200 lines
7.7 KiB
Python
200 lines
7.7 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
|
|
SHA256_RE = re.compile(r'^[0-9a-f]{64}$')
|
|
DOCKER_DIGEST_RE = re.compile(r'^sha256:[0-9a-f]{64}$')
|
|
DOCKER_TAG_RE = r'[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}'
|
|
DOCKER_NAME_COMPONENT_RE = r'[a-z0-9]+(?:(?:[._]|__|[-]+)[a-z0-9]+)*'
|
|
DOCKER_DOMAIN_COMPONENT_RE = r'[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?'
|
|
DOCKER_PORT_RE = r'[0-9]{1,5}'
|
|
DOCKER_REGISTRY_RE = (
|
|
rf'(?:\[[0-9A-Fa-f:.]+\](?::{DOCKER_PORT_RE})?'
|
|
rf'|(?:localhost|{DOCKER_DOMAIN_COMPONENT_RE}(?:\.{DOCKER_DOMAIN_COMPONENT_RE})*)(?::{DOCKER_PORT_RE})?)'
|
|
)
|
|
DOCKER_IMAGE_RE = re.compile(
|
|
rf'^(?=.{{1,512}}$)(?:{DOCKER_REGISTRY_RE}/)?'
|
|
rf'{DOCKER_NAME_COMPONENT_RE}(?:/{DOCKER_NAME_COMPONENT_RE})*'
|
|
rf'(?::{DOCKER_TAG_RE})?(?:@sha256:[0-9a-f]{{64}})?$'
|
|
)
|
|
DOCKER_REVISION_RE = re.compile(r'^[A-Za-z0-9:+._-]{1,128}$')
|
|
DOCKER_TAG_TARGET_SCHEMA = 'docker-tag-v1'
|
|
DOCKERHUB_REGISTRIES = frozenset((
|
|
'docker.io', 'index.docker.io', 'registry-1.docker.io',
|
|
))
|
|
HUGGINGFACE_SPACE_COMPONENT_RE = re.compile(
|
|
r'^[A-Za-z0-9_](?:[A-Za-z0-9._-]{0,94}[A-Za-z0-9_])?$'
|
|
)
|
|
|
|
|
|
def _postman_data(target):
|
|
if isinstance(target, dict):
|
|
return dict(target)
|
|
text = str(target or '').strip()
|
|
if not text:
|
|
return {}
|
|
if text.startswith('{'):
|
|
value = json.loads(text)
|
|
if not isinstance(value, dict):
|
|
raise ValueError('Postman target JSON must be an object')
|
|
return value
|
|
if text.lower().startswith('file:'):
|
|
return {'source': 'local_file', 'local_path': text[5:]}
|
|
if '://' in text:
|
|
return {'source': 'url', 'url': text}
|
|
return {'source': 'local_file', 'local_path': text}
|
|
|
|
|
|
def postman_target_identity(target):
|
|
"""Return artifact identity independent of discovery-origin metadata."""
|
|
try:
|
|
data = _postman_data(target)
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
return str(target or '').strip().lower()
|
|
digest = str(data.get('sha256') or data.get('hash') or data.get('cache_key') or '').strip().lower()
|
|
if SHA256_RE.fullmatch(digest):
|
|
return f'postman:sha256:{digest}'
|
|
cache_path = str(data.get('cache_path') or '')
|
|
cache_name = os.path.basename(cache_path).lower()
|
|
cache_digest = cache_name.split('.', 1)[0]
|
|
if SHA256_RE.fullmatch(cache_digest):
|
|
return f'postman:sha256:{cache_digest}'
|
|
source = str(data.get('source') or '').lower()
|
|
if source == 'github_code':
|
|
return 'postman:github_code:{}:{}:{}'.format(
|
|
str(data.get('repo') or '').lower(),
|
|
str(data.get('path') or '').replace('\\', '/').lower(),
|
|
str(data.get('sha') or '').lower(),
|
|
)
|
|
if source == 'url':
|
|
return f'postman:url:{str(data.get("url") or "").rstrip("/").lower()}'
|
|
local_path = data.get('local_path') or data.get('path') or cache_path
|
|
if local_path:
|
|
canonical = os.path.normcase(os.path.realpath(os.path.abspath(os.fspath(local_path))))
|
|
return f'postman:file:{canonical}'
|
|
stable = {
|
|
key: value for key, value in data.items()
|
|
if key not in {'origin', 'cached_at', 'discovered_at', 'size', 'bytes'}
|
|
}
|
|
payload = json.dumps(stable, ensure_ascii=True, sort_keys=True, separators=(',', ':'), default=str)
|
|
return 'postman:metadata:' + hashlib.sha256(payload.encode('utf-8')).hexdigest()
|
|
|
|
|
|
def normalize_docker_digest(value):
|
|
digest = str(value or '').strip().lower()
|
|
return digest if DOCKER_DIGEST_RE.fullmatch(digest) else ''
|
|
|
|
|
|
def validate_docker_image_reference(value, require_digest=True):
|
|
image = str(value or '').strip()
|
|
if not image or image.startswith('{') or '://' in image or not DOCKER_IMAGE_RE.fullmatch(image):
|
|
raise ValueError('invalid Docker image reference')
|
|
digest = image.rsplit('@', 1)[1] if '@' in image else ''
|
|
if require_digest and not normalize_docker_digest(digest):
|
|
raise ValueError('Docker image reference must include a sha256 digest')
|
|
return image
|
|
|
|
|
|
def serialize_docker_tag_target(image, revision):
|
|
image = validate_docker_image_reference(image)
|
|
revision = str(revision or '').strip()
|
|
if not DOCKER_REVISION_RE.fullmatch(revision):
|
|
raise ValueError('invalid Docker tag revision')
|
|
return json.dumps(
|
|
{
|
|
'image': image,
|
|
'revision': revision,
|
|
'schema': DOCKER_TAG_TARGET_SCHEMA,
|
|
},
|
|
ensure_ascii=True,
|
|
sort_keys=True,
|
|
separators=(',', ':'),
|
|
)
|
|
|
|
|
|
def parse_docker_target(target):
|
|
structured = isinstance(target, dict)
|
|
if structured:
|
|
data = dict(target)
|
|
else:
|
|
text = str(target or '').strip()
|
|
if not text.startswith('{'):
|
|
image = validate_docker_image_reference(text)
|
|
return {'image': image, 'revision': '', 'structured': False, 'target': image}
|
|
try:
|
|
data = json.loads(text)
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError('invalid Docker target JSON') from exc
|
|
if not isinstance(data, dict):
|
|
raise ValueError('Docker target JSON must be an object')
|
|
structured = True
|
|
|
|
if set(data) != {'schema', 'image', 'revision'} or data.get('schema') != DOCKER_TAG_TARGET_SCHEMA:
|
|
raise ValueError('unsupported Docker target structure')
|
|
canonical = serialize_docker_tag_target(data.get('image'), data.get('revision'))
|
|
return {
|
|
'image': validate_docker_image_reference(data.get('image')),
|
|
'revision': str(data.get('revision')).strip(),
|
|
'structured': structured,
|
|
'target': canonical,
|
|
}
|
|
|
|
|
|
def parse_dockerhub_digest_target(target):
|
|
"""Return canonical parts for a public immutable DockerHub image target."""
|
|
parsed = parse_docker_target(target)
|
|
image = parsed['image']
|
|
if image != image.lower():
|
|
raise ValueError('DockerHub image reference must be canonical lowercase')
|
|
image_name, separator, digest = image.rpartition('@')
|
|
if not separator or not normalize_docker_digest(digest):
|
|
raise ValueError('DockerHub image reference must include a sha256 digest')
|
|
repository_with_registry = image_name.rsplit(':', 1)[0] if ':' in image_name.rsplit('/', 1)[-1] else image_name
|
|
parts = repository_with_registry.split('/')
|
|
first = parts[0]
|
|
has_registry = first == 'localhost' or '.' in first or ':' in first or first.startswith('[')
|
|
if has_registry:
|
|
registry = first
|
|
if registry not in DOCKERHUB_REGISTRIES or len(parts) < 2:
|
|
raise ValueError('Docker direct execution requires a public DockerHub image')
|
|
repository = '/'.join(parts[1:])
|
|
else:
|
|
registry = 'docker.io'
|
|
repository = repository_with_registry
|
|
if '/' not in repository:
|
|
repository = 'library/' + repository
|
|
return {
|
|
**parsed,
|
|
'image': image,
|
|
'registry': registry,
|
|
'repository': repository,
|
|
'manifest_digest': digest,
|
|
'normalized_target': image,
|
|
}
|
|
|
|
|
|
def normalize_huggingface_space_id(value):
|
|
"""Validate and return one canonical case-preserved HuggingFace Space ID."""
|
|
if not isinstance(value, str) or value != value.strip() or not 3 <= len(value) <= 96:
|
|
raise ValueError('invalid HuggingFace Space identifier')
|
|
if value.count('/') != 1 or '\\' in value or '..' in value or '--' in value:
|
|
raise ValueError('invalid HuggingFace Space identifier')
|
|
namespace, name = value.split('/', 1)
|
|
if (
|
|
HUGGINGFACE_SPACE_COMPONENT_RE.fullmatch(namespace) is None
|
|
or HUGGINGFACE_SPACE_COMPONENT_RE.fullmatch(name) is None
|
|
or name.lower().endswith('.git')
|
|
):
|
|
raise ValueError('invalid HuggingFace Space identifier')
|
|
return value
|
|
|
|
|
|
def docker_target_identity(target):
|
|
try:
|
|
data = parse_docker_target(target)
|
|
except (TypeError, ValueError):
|
|
return str(target or '').strip().lower()
|
|
return data['image'].lower()
|