mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
00d7816cfd
A submodule on the repo's host was allowed unpinned, on any port. The owner of the repo's domain could serve the clone from a public address, then answer the submodule's lookup with an internal address (DNS rebinding), or point it at another port on that name. The repo's host is now resolved once per download, and every fetch (ls-remote, clone, fetch, submodule update) is pinned to that answer. A submodule on the repo's host *and port* is fetched from those same addresses, still allowing a LAN forge the user chose; another port on that name is checked like any other host. Tests no longer use real DNS. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
945 lines
39 KiB
Python
945 lines
39 KiB
Python
import asyncio
|
|
import html
|
|
import ipaddress
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import stat
|
|
from dataclasses import dataclass, field
|
|
from typing import List, NamedTuple, Optional, Tuple
|
|
from urllib.parse import urlparse
|
|
|
|
from sqlalchemy import func, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from wrolpi import fts
|
|
from wrolpi.cmd import GIT_BIN, run_command
|
|
from wrolpi.collections.models import Collection
|
|
from wrolpi.common import ConfigFile, get_media_directory, get_relative_to_media_directory, logger, \
|
|
escape_file_name, background_task
|
|
from wrolpi.db import get_db_session
|
|
from wrolpi.downloader import DownloadFrequency, save_downloads_config
|
|
from wrolpi.errors import UnknownDirectory, ValidationError
|
|
from wrolpi.events import Events
|
|
from wrolpi.switches import register_switch_handler, ActivateSwitchMethod
|
|
from wrolpi.tags import Tag
|
|
from wrolpi.vars import FILE_MAX_TEXT_SIZE
|
|
from .errors import InvalidRepo, RepoConflict
|
|
from .models import Repository, REPO_MODES
|
|
|
|
logger = logger.getChild(__name__)
|
|
|
|
__all__ = ['ParsedRepoURL', 'parse_repo_url', 'repo_url_key', 'git_command', 'git_env', 'create_repository',
|
|
'update_repository', 'delete_repository', 'get_repositories', 'list_repo_tree', 'find_readme',
|
|
'repo_destination_variables', 'is_valid_branch', 'submodule_url_error', 'HostAddresses',
|
|
'resolve_repo_addresses', 'pin_config', 'is_public_address', 'submodule_fetch_config', 'read_readme_text',
|
|
'search_repos', 'count_repos', 'search_repos_by_name', 'get_repo_log', 'get_repo_root',
|
|
'repo_archive_name', 'stream_repo_archive', 'ReposConfig', 'repos_config', 'get_repos_config',
|
|
'save_repos_config', 'import_repos_config', 'DEFAULT_REPO_FREQUENCY']
|
|
|
|
DEFAULT_REPO_FREQUENCY = DownloadFrequency.weekly.value
|
|
|
|
# The only transports git may use. Submodules and redirects cannot reach `file://`, `ext::`, ssh, etc.
|
|
ALLOWED_PROTOCOLS = ('https',)
|
|
|
|
# README files in the order they are preferred; matched case-insensitively.
|
|
README_NAMES = ('readme.md', 'readme.markdown', 'readme.rst', 'readme.txt', 'readme')
|
|
|
|
# A forge's web URL of a branch: GitHub `/owner/repo/tree/<branch>`, GitLab `/group/repo/-/tree/<branch>`.
|
|
TREE_URL_RE = re.compile(r'^(?P<repo>/.+?)(?:/-)?/tree/(?P<branch>[^?#]+)$')
|
|
|
|
|
|
@dataclass
|
|
class ParsedRepoURL:
|
|
url: str
|
|
host: str
|
|
owner: str
|
|
name: str
|
|
branch: Optional[str] = None
|
|
|
|
|
|
def parse_repo_url(url: str) -> ParsedRepoURL:
|
|
"""Validate an https git URL and split it into its parts.
|
|
|
|
A forge's branch URL (e.g. `https://github.com/owner/repo/tree/dev`) becomes the repo URL and its branch."""
|
|
url = (url or '').strip()
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ALLOWED_PROTOCOLS:
|
|
raise InvalidRepo('Only https:// repo URLs are supported')
|
|
if parsed.username or parsed.password:
|
|
# Credentials would be saved in the configs.
|
|
raise InvalidRepo('Repo URL must not contain a username or password')
|
|
if not parsed.hostname:
|
|
raise InvalidRepo('Repo URL must have a host')
|
|
if parsed.query or parsed.fragment:
|
|
raise InvalidRepo('Repo URL must not contain a query or fragment')
|
|
|
|
path = parsed.path.rstrip('/')
|
|
branch = None
|
|
if match := TREE_URL_RE.match(path):
|
|
path, branch = match.group('repo'), match.group('branch')
|
|
|
|
parts = [i for i in path.split('/') if i]
|
|
if not parts:
|
|
raise InvalidRepo('Repo URL must have a path')
|
|
if any(i in ('.', '..') for i in parts):
|
|
# The name and owner are used in the Repo's directory.
|
|
raise InvalidRepo('Repo URL must not contain "." or ".." in its path')
|
|
name = parts[-1].removesuffix('.git')
|
|
if not name:
|
|
raise InvalidRepo('Repo URL must have a name')
|
|
owner = '/'.join(parts[:-1])
|
|
|
|
host = parsed.hostname.lower()
|
|
netloc = f'{host}:{parsed.port}' if parsed.port else host
|
|
return ParsedRepoURL(url=f'{parsed.scheme}://{netloc}{path}', host=host, owner=owner, name=name, branch=branch)
|
|
|
|
|
|
# Characters git forbids in a ref name (`git check-ref-format`), plus whitespace.
|
|
INVALID_BRANCH_CHARACTERS_RE = re.compile(r'[\x00-\x20\x7f~^:?*\[\\]')
|
|
|
|
|
|
def is_valid_branch(branch: str) -> bool:
|
|
"""Is `branch` a valid git branch name (the rules of `git check-ref-format --branch`)?
|
|
|
|
Branch names are passed to git as arguments, so a name starting with `-` (an option) is never valid."""
|
|
if not branch or not isinstance(branch, str):
|
|
return False
|
|
if branch.startswith('-') or branch == '@' or '@{' in branch or '..' in branch:
|
|
return False
|
|
if INVALID_BRANCH_CHARACTERS_RE.search(branch):
|
|
return False
|
|
if branch.startswith('/') or branch.endswith('/') or '//' in branch or branch.endswith('.'):
|
|
return False
|
|
for component in branch.split('/'):
|
|
if component.startswith('.') or component.endswith('.lock'):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _validate_branch(branch: Optional[str]):
|
|
if branch is not None and not is_valid_branch(branch):
|
|
raise ValidationError(f'Invalid branch name: {branch!r}')
|
|
|
|
|
|
def resolve_host_addresses(host: str) -> List[str]:
|
|
"""The IP addresses `host` resolves to."""
|
|
return list({i[4][0] for i in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM)})
|
|
|
|
|
|
# IPv6 ranges which carry an IPv4 address, or are not public.
|
|
NAT64_NETWORK = ipaddress.ip_network('64:ff9b::/96')
|
|
LOCAL_NAT64_NETWORK = ipaddress.ip_network('64:ff9b:1::/48')
|
|
TEREDO_NETWORK = ipaddress.ip_network('2001::/32')
|
|
SITE_LOCAL_NETWORK = ipaddress.ip_network('fec0::/10')
|
|
|
|
|
|
def is_public_address(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
|
|
"""Is `address` a public (globally routed) unicast address?
|
|
|
|
`is_global` alone is not enough: it accepts multicast, deprecated site-local, and IPv6 addresses which carry a
|
|
private IPv4 address (e.g. NAT64 `64:ff9b::169.254.169.254`, the metadata address)."""
|
|
if isinstance(address, ipaddress.IPv6Address):
|
|
if address.ipv4_mapped:
|
|
return is_public_address(address.ipv4_mapped)
|
|
if address in NAT64_NETWORK:
|
|
return is_public_address(ipaddress.IPv4Address(int(address) & 0xFFFFFFFF))
|
|
if address.sixtofour:
|
|
return is_public_address(address.sixtofour)
|
|
if address in LOCAL_NAT64_NETWORK or address in TEREDO_NETWORK or address in SITE_LOCAL_NETWORK:
|
|
return False
|
|
return address.is_global and not (
|
|
address.is_multicast or address.is_reserved or address.is_link_local or address.is_loopback
|
|
or address.is_private or address.is_unspecified)
|
|
|
|
|
|
# Submodule URLs must have a shape that every URL parser (Python's, git's, curl's) reads the same way: no userinfo,
|
|
# backslashes, queries, fragments or whitespace.
|
|
ABSOLUTE_SUBMODULE_URL_RE = re.compile(
|
|
r'^https://(?P<host>[A-Za-z0-9.-]+|\[[0-9A-Fa-f:.]+\])'
|
|
r'(?::(?P<port>[0-9]{1,5}))?'
|
|
r'(?:/[A-Za-z0-9._~%!$&\'()*+,;=:/-]*)?$')
|
|
|
|
|
|
class HostAddresses(NamedTuple):
|
|
"""A host and port, and the addresses its name resolved to (None for an IP address, or a failed lookup)."""
|
|
host: str
|
|
port: int
|
|
addresses: Optional[List[str]]
|
|
|
|
|
|
def resolve_repo_addresses(url: str) -> HostAddresses:
|
|
"""Resolve the host of a Repo's URL, once per download. Every fetch from the host is pinned to the result."""
|
|
parsed = urlparse(url)
|
|
host = (parsed.hostname or '').lower()
|
|
port = parsed.port or 443
|
|
if _is_ip_address(host):
|
|
return HostAddresses(host, port, None)
|
|
try:
|
|
return HostAddresses(host, port, sorted(resolve_host_addresses(host)))
|
|
except OSError:
|
|
return HostAddresses(host, port, None)
|
|
|
|
|
|
def pin_config(pin: HostAddresses) -> Optional[str]:
|
|
"""git config which makes git connect to `pin.host` at exactly `pin.addresses`."""
|
|
if not pin.addresses:
|
|
return None
|
|
addresses = ','.join(f'[{i}]' if ':' in i else i for i in pin.addresses)
|
|
return f'http.curloptResolve={pin.host}:{pin.port}:{addresses}'
|
|
|
|
|
|
def _is_ip_address(host: str) -> bool:
|
|
try:
|
|
ipaddress.ip_address(host.strip('[]'))
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _check_submodule_url(url: str, repo: HostAddresses) -> Tuple[Optional[str], Optional[HostAddresses]]:
|
|
"""Returns why the URL must not be fetched (or None), and the host to pin it to (or None).
|
|
|
|
`url` must be absolute: git resolves relative submodule URLs (which can climb to another host, e.g.
|
|
`../../../10.0.0.1/x`), and the result is checked."""
|
|
match = ABSOLUTE_SUBMODULE_URL_RE.match(url)
|
|
if not match:
|
|
return 'only plain https:// submodule URLs can be downloaded', None
|
|
host = match.group('host').lower()
|
|
port = int(match.group('port') or 443)
|
|
if (host, port) == (repo.host, repo.port):
|
|
# The host the user chose (it may be on their network). It is fetched from the addresses the repo was
|
|
# fetched from (`pin_config(repo)` is part of every repo fetch), so it is not resolved again.
|
|
if repo.addresses or _is_ip_address(host):
|
|
return None, None
|
|
return f'could not resolve {host}', None
|
|
if _is_ip_address(host):
|
|
addresses = [ipaddress.ip_address(host.strip('[]'))]
|
|
pin = None # An address needs no DNS.
|
|
else:
|
|
try:
|
|
addresses = [ipaddress.ip_address(i.split('%')[0]) for i in resolve_host_addresses(host)]
|
|
except (OSError, ValueError):
|
|
return f'could not resolve {host}', None
|
|
pin = HostAddresses(host, port, [str(i) for i in addresses])
|
|
if not addresses or not all(is_public_address(i) for i in addresses):
|
|
return f'{host} is not a public address', None
|
|
return None, pin
|
|
|
|
|
|
def submodule_url_error(url: str, repo: HostAddresses) -> Optional[str]:
|
|
"""Why a submodule URL must not be fetched, or None when it may be.
|
|
|
|
A submodule's URL is chosen by whoever controls the repo, so it may not make WROLPi fetch from its own network
|
|
(e.g. `https://localhost:8443/...` or `https://10.0.0.1/...`). Allowed: the repo's own host and port (which
|
|
the user chose), and hosts with only public addresses. A relative URL must first be resolved by git."""
|
|
return _check_submodule_url(url, repo)[0]
|
|
|
|
|
|
def submodule_fetch_config(urls: List[str], repo: HostAddresses) -> tuple:
|
|
"""git config for fetching submodules with these URLs. Raises ValueError if any URL must not be fetched.
|
|
|
|
Each checked host is pinned to the addresses that were checked, so its DNS cannot change before git fetches it,
|
|
and redirects are not followed, so a checked host cannot send git elsewhere. (The repo's own host is pinned by
|
|
`pin_config(repo)`.)"""
|
|
config = ['http.followRedirects=false']
|
|
pinned = set()
|
|
for url in urls:
|
|
error, pin = _check_submodule_url(url, repo)
|
|
if error:
|
|
raise ValueError(f'Refusing to download submodule {url}: {error}')
|
|
if pin and (pin.host, pin.port) not in pinned:
|
|
pinned.add((pin.host, pin.port))
|
|
config.append(pin_config(pin))
|
|
return tuple(config)
|
|
|
|
|
|
def repo_url_key(url: str) -> str:
|
|
"""Two URLs of the same repo have the same key (`.git` suffix and case of the host are ignored)."""
|
|
parsed = urlparse(url.strip())
|
|
return f'{(parsed.hostname or "").lower()}{parsed.path.rstrip("/").removesuffix(".git")}'
|
|
|
|
|
|
def git_command(*args, config: tuple = ()) -> tuple:
|
|
"""Build a git command that is safe to run against an untrusted remote repo."""
|
|
if not GIT_BIN:
|
|
raise InvalidRepo('git is not installed')
|
|
safety = [
|
|
'-c', 'protocol.allow=never',
|
|
*(i for p in ALLOWED_PROTOCOLS for i in ('-c', f'protocol.{p}.allow=always')),
|
|
# Never run the repo's hooks.
|
|
'-c', 'core.hooksPath=/dev/null',
|
|
# A symlink in a repo could point outside the repo (e.g. /etc/shadow), and media files are served.
|
|
# Symlinks are checked out as plain files containing the link's target.
|
|
'-c', 'core.symlinks=false',
|
|
'-c', 'core.fsmonitor=false',
|
|
# The media directory may be owned by another user (e.g. a Docker volume).
|
|
'-c', 'safe.directory=*',
|
|
'-c', 'credential.helper=',
|
|
'-c', 'submodule.recurse=false',
|
|
]
|
|
for i in config:
|
|
safety.extend(['-c', i])
|
|
return GIT_BIN, *safety, *args
|
|
|
|
|
|
# The only parts of WROLPi's environment git may see. Anything else (GIT_ALLOW_PROTOCOL, GIT_CONFIG_COUNT and
|
|
# GIT_CONFIG_KEY_*/VALUE_*, GIT_CONFIG_PARAMETERS, GIT_SSL_NO_VERIFY, proxies, ...) could undo `git_command`'s
|
|
# restrictions.
|
|
GIT_ENV_ALLOWED = ('PATH', 'HOME', 'TMPDIR', 'TZ', 'SSL_CERT_FILE', 'SSL_CERT_DIR')
|
|
|
|
|
|
def git_env() -> dict:
|
|
"""Environment for git: never prompt for credentials, ignore the user's and system's git configs, and inherit
|
|
nothing that changes how git connects."""
|
|
env = {k: v for k, v in os.environ.items() if k in GIT_ENV_ALLOWED}
|
|
env.update(
|
|
GIT_TERMINAL_PROMPT='0',
|
|
GIT_ASKPASS='true',
|
|
SSH_ASKPASS='true',
|
|
GIT_CONFIG_NOSYSTEM='1',
|
|
GIT_CONFIG_GLOBAL=os.devnull,
|
|
GIT_LFS_SKIP_SMUDGE='1',
|
|
LC_ALL='C',
|
|
)
|
|
return env
|
|
|
|
|
|
def find_readme(directory: pathlib.Path) -> Optional[str]:
|
|
"""Return the name of the preferred README file in `directory`, if any."""
|
|
try:
|
|
files = {i.name.lower(): i.name for i in directory.iterdir() if i.is_file() and not i.is_symlink()}
|
|
except FileNotFoundError:
|
|
return None
|
|
for name in README_NAMES:
|
|
if name in files:
|
|
return files[name]
|
|
return None
|
|
|
|
|
|
MARKDOWN_SUFFIXES = ('.md', '.markdown')
|
|
# A README is untrusted; every pattern here must run in linear time. Each bracketed run stops at the next
|
|
# opening bracket, so scans from different starting points never overlap.
|
|
INLINE_MARKDOWN_PATTERNS = [
|
|
(re.compile(r'<[^<>\n]*>'), ' '), # HTML tags.
|
|
(re.compile(r'!\[([^\[\]\n]*)\]\([^()\n]*\)'), r'\1'), # Images (and badges) become their alt text.
|
|
(re.compile(r'\[([^\[\]\n]*)\]\([^()\n]*\)'), r'\1'), # Links become their text.
|
|
(re.compile(r'[*`|]+'), ' '), # Emphasis, inline code, table cells.
|
|
(re.compile(r'[ \t]+'), ' '),
|
|
]
|
|
HEADING_RE = re.compile(r'#{1,6}(?=\s|$)')
|
|
LINK_REFERENCE_RE = re.compile(r'\[[^\[\]]+\]:\s')
|
|
TABLE_SEPARATOR_CHARACTERS = set('|:- \t')
|
|
|
|
|
|
def _remove_html_comments(text: str) -> str:
|
|
parts, position = [], 0
|
|
while (comment := text.find('<!--', position)) != -1:
|
|
parts.append(text[position:comment])
|
|
end = text.find('-->', comment + 4)
|
|
if end == -1:
|
|
return ''.join(parts)
|
|
parts.append(' ')
|
|
position = end + 3
|
|
parts.append(text[position:])
|
|
return ''.join(parts)
|
|
|
|
|
|
def markdown_to_text(text: str) -> str:
|
|
"""Remove markdown syntax so search snippets read as plain text."""
|
|
lines = []
|
|
for line in _remove_html_comments(text).splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith(('```', '~~~')) or LINK_REFERENCE_RE.match(stripped):
|
|
continue
|
|
if '-' in stripped and set(stripped) <= TABLE_SEPARATOR_CHARACTERS:
|
|
continue # A table separator row, or a horizontal rule.
|
|
stripped = stripped.lstrip('>').strip()
|
|
if heading := HEADING_RE.match(stripped):
|
|
stripped = stripped[heading.end():].strip()
|
|
for pattern, replacement in INLINE_MARKDOWN_PATTERNS:
|
|
stripped = pattern.sub(replacement, stripped)
|
|
if stripped := stripped.strip():
|
|
lines.append(stripped)
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def read_readme_text(path: pathlib.Path) -> Optional[str]:
|
|
"""Read a README as plain text for search; markdown syntax is removed so snippets read well.
|
|
|
|
A symlink is never followed (it could point outside the repo)."""
|
|
try:
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
|
|
except OSError:
|
|
return None
|
|
with os.fdopen(fd, 'rb') as fh:
|
|
if not stat.S_ISREG(os.fstat(fh.fileno()).st_mode):
|
|
return None
|
|
text = fh.read(FILE_MAX_TEXT_SIZE).decode(errors='replace')
|
|
if path.suffix.lower() in MARKDOWN_SUFFIXES:
|
|
text = markdown_to_text(text)
|
|
return text.strip() or None
|
|
|
|
|
|
def repo_destination_variables(collection: Collection) -> dict:
|
|
"""Variables for `repos_destination` which are stored on the Repository (not the Collection)."""
|
|
session = Session.object_session(collection)
|
|
repo = Repository.get_by_collection_id(session, collection.id) if session and collection.id else None
|
|
return dict(
|
|
repo_owner=escape_file_name(repo.owner) if repo and repo.owner else '',
|
|
repo_host=escape_file_name(repo.host) if repo and repo.host else '',
|
|
)
|
|
|
|
|
|
def get_repository_by_url(session: Session, url: str) -> Optional[Repository]:
|
|
key = repo_url_key(url)
|
|
for id_, repo_url in session.query(Repository.id, Repository.url):
|
|
if repo_url_key(repo_url) == key:
|
|
return session.query(Repository).filter_by(id=id_).one()
|
|
return None
|
|
|
|
|
|
def _validate_mode(mode: str):
|
|
if mode not in REPO_MODES:
|
|
raise ValidationError(f'Repo mode must be one of {", ".join(REPO_MODES)}')
|
|
|
|
|
|
def _validate_directory(directory: pathlib.Path):
|
|
"""A Repo's directory is strictly inside the media directory; it is cloned into, and may be deleted."""
|
|
media_directory = get_media_directory()
|
|
directory = pathlib.Path(directory)
|
|
if '..' in directory.parts or not directory.is_absolute():
|
|
raise InvalidRepo(f'Invalid repo directory: {directory}')
|
|
try:
|
|
relative = directory.relative_to(media_directory)
|
|
except ValueError:
|
|
raise InvalidRepo('Repo directory must be in the media directory')
|
|
if not relative.parts:
|
|
raise InvalidRepo('Repo directory can not be the media directory')
|
|
|
|
|
|
def _validate_frequency(frequency: int):
|
|
if not isinstance(frequency, int) or frequency <= 0:
|
|
raise ValidationError('Repo frequency must be a positive number of seconds')
|
|
|
|
|
|
def create_repository(session: Session, url: str, tag_name: str = None, frequency: int = DEFAULT_REPO_FREQUENCY,
|
|
mode: str = 'full', branch: str = None, name: str = None, description: str = None,
|
|
directory: pathlib.Path = None, submodules: bool = False) -> Repository:
|
|
"""Create a Repo (Collection + Repository + recurring Download). The download clones it."""
|
|
parsed = parse_repo_url(url)
|
|
_validate_mode(mode)
|
|
_validate_frequency(frequency)
|
|
branch = branch or parsed.branch or None
|
|
_validate_branch(branch)
|
|
name = (name or parsed.name).strip()
|
|
if name in ('', '.', '..'):
|
|
raise InvalidRepo(f'Invalid repo name: {name!r}')
|
|
|
|
if get_repository_by_url(session, parsed.url):
|
|
raise RepoConflict(f'A repo with this URL already exists: {parsed.url}')
|
|
if session.query(Collection).filter_by(kind='repo', name=name).one_or_none():
|
|
suggestion = f'{parsed.owner.replace("/", "-")}-{name}' if parsed.owner else f'{parsed.host}-{name}'
|
|
raise RepoConflict(f'A repo named {name!r} already exists. Try the name {suggestion!r}')
|
|
|
|
tag = Tag.get_or_create_tag(session, tag_name) if tag_name else None
|
|
collection = Collection(kind='repo', name=name, description=description, tag=tag)
|
|
session.add(collection)
|
|
session.flush([collection])
|
|
repo = Repository(url=parsed.url, host=parsed.host, owner=parsed.owner, mode=mode, branch=branch,
|
|
submodules=bool(submodules), collection=collection)
|
|
repo.update_search_name()
|
|
session.add(repo)
|
|
session.flush([repo])
|
|
|
|
directory = directory or collection.format_destination(tag_name)
|
|
_validate_directory(directory)
|
|
if session.query(Collection).filter(Collection.directory == directory, Collection.id != collection.id).first():
|
|
relative = get_relative_to_media_directory(directory)
|
|
raise RepoConflict(f'Another collection already uses the directory {relative}')
|
|
collection.directory = directory
|
|
|
|
download = collection.get_or_create_download(session, parsed.url, frequency, downloader_name='git')
|
|
# The Download may already exist (e.g. imported from download_manager.yaml).
|
|
download.frequency = frequency
|
|
session.commit()
|
|
|
|
save_repos_config.activate_switch()
|
|
save_downloads_config.activate_switch()
|
|
return repo
|
|
|
|
|
|
def update_repository(session: Session, repo_id: int, description: str = None, frequency: int = None,
|
|
mode: str = None, branch: str = None, submodules: bool = None) -> Repository:
|
|
"""Change a Repo's settings. An empty `branch` follows the remote's default branch."""
|
|
repo = Repository.find_by_id(session, repo_id)
|
|
if description is not None:
|
|
repo.collection.description = description or None
|
|
if mode is not None:
|
|
_validate_mode(mode)
|
|
repo.mode = mode
|
|
if branch is not None:
|
|
branch = branch.strip() or None
|
|
_validate_branch(branch)
|
|
repo.branch = branch
|
|
if submodules is not None:
|
|
repo.submodules = bool(submodules)
|
|
if frequency is not None:
|
|
_validate_frequency(frequency)
|
|
if download := repo.download:
|
|
download.frequency = frequency
|
|
else:
|
|
repo.collection.get_or_create_download(session, repo.url, frequency, downloader_name='git')
|
|
session.commit()
|
|
|
|
save_repos_config.activate_switch()
|
|
save_downloads_config.activate_switch()
|
|
return repo
|
|
|
|
|
|
def _delete_repository(session: Session, repo: Repository):
|
|
collection = repo.collection
|
|
for download in list(collection.downloads):
|
|
session.delete(download)
|
|
session.delete(collection)
|
|
session.flush()
|
|
|
|
|
|
async def delete_repository(session: Session, repo_id: int, delete_files: bool = False) -> dict:
|
|
"""Delete a Repo and its Download. The repo's files are kept unless `delete_files`."""
|
|
repo = Repository.find_by_id(session, repo_id)
|
|
directory = repo.directory
|
|
result = dict(id=repo.id, name=repo.name, directory=str(get_relative_to_media_directory(directory)))
|
|
_delete_repository(session, repo)
|
|
session.commit()
|
|
|
|
if delete_files and directory and directory.is_dir():
|
|
media_directory = get_media_directory().resolve()
|
|
if directory.resolve() == media_directory or media_directory not in directory.resolve().parents:
|
|
raise InvalidRepo('Refusing to delete a directory outside the media directory')
|
|
background_task(_delete_directory(directory))
|
|
|
|
save_repos_config.activate_switch()
|
|
save_downloads_config.activate_switch()
|
|
return result
|
|
|
|
|
|
async def _delete_directory(directory: pathlib.Path):
|
|
await asyncio.to_thread(shutil.rmtree, directory)
|
|
logger.info(f'Deleted repo directory {directory}')
|
|
|
|
|
|
def get_repositories(session: Session) -> List[dict]:
|
|
repos = session.query(Repository).options(*Repository.json_options()) \
|
|
.join(Collection, Collection.id == Repository.collection_id) \
|
|
.order_by(func.lower(Collection.name)).all()
|
|
return [i.__json__() for i in repos]
|
|
|
|
|
|
def _repo_search_sql(search_str: Optional[str], tag_names: Optional[List[str]]) \
|
|
-> Optional[Tuple[str, str, dict, bool]]:
|
|
"""The FROM and WHERE of a Repo search by name/README and Tag, its params, and whether it uses FTS.
|
|
|
|
Returns None when nothing can match."""
|
|
tag_names = tag_names or []
|
|
if len(tag_names) > 1:
|
|
# A Repo has one Tag.
|
|
return None
|
|
search = fts.repository_search_join(search_str, start=HIGHLIGHT_START, stop=HIGHLIGHT_STOP) \
|
|
if search_str else None
|
|
if search is None and (search_str and search_str.strip() or not tag_names):
|
|
# Nothing usable to search, or nothing to filter by.
|
|
return None
|
|
|
|
from_sql = 'repository r JOIN collection c ON c.id = r.collection_id'
|
|
where, params = [], {}
|
|
if search:
|
|
from_sql = f'{from_sql} {search.join}'
|
|
params.update(search.params)
|
|
if tag_names:
|
|
from_sql = f'{from_sql} JOIN tag t ON t.id = c.tag_id'
|
|
where.append('t.name = :tag_name')
|
|
params['tag_name'] = tag_names[0]
|
|
return from_sql, (f"WHERE {' AND '.join(where)}" if where else ''), params, search is not None
|
|
|
|
|
|
def search_repos(session: Session, search_str: str = None, tag_names: List[str] = None, limit: int = 20,
|
|
offset: int = 0) -> Tuple[List[dict], int]:
|
|
"""Find Repos by their name and README (and/or their Tag). Returns the Repos (best match first, each with a
|
|
`readme_headline`) and the total count."""
|
|
query = _repo_search_sql(search_str, tag_names)
|
|
if query is None:
|
|
return [], 0
|
|
from_sql, where, params, is_fts = query
|
|
select = 'r.id, fts.readme_headline' if is_fts else 'r.id, NULL'
|
|
order_by = 'fts.ts_rank DESC, LOWER(c.name)' if is_fts else 'LOWER(c.name)'
|
|
rows = session.execute(text(
|
|
f'SELECT {select}, COUNT(*) OVER() FROM {from_sql} {where} ORDER BY {order_by} LIMIT :limit OFFSET :offset'
|
|
), dict(params, limit=limit, offset=offset)).fetchall()
|
|
if not rows:
|
|
return [], 0
|
|
|
|
repos = session.query(Repository).options(*Repository.json_options()) \
|
|
.filter(Repository.id.in_([i[0] for i in rows])).all()
|
|
repos = {i.id: i for i in repos}
|
|
results = [dict(repos[id_].__json__(), readme_headline=_safe_headline(headline))
|
|
for id_, headline, _ in rows if id_ in repos]
|
|
return results, rows[0][2]
|
|
|
|
|
|
# Highlight markers which cannot be confused with HTML. See `_safe_headline`.
|
|
HIGHLIGHT_START, HIGHLIGHT_STOP = '\x02', '\x03'
|
|
|
|
|
|
def _safe_headline(headline: Optional[str]) -> Optional[str]:
|
|
"""A README is untrusted and its headline is rendered as HTML: escape it, then mark the matches with <b>."""
|
|
if not headline:
|
|
return headline
|
|
return html.escape(headline).replace(HIGHLIGHT_START, '<b>').replace(HIGHLIGHT_STOP, '</b>')
|
|
|
|
|
|
def count_repos(session: Session, search_str: str = None, tag_names: List[str] = None) -> int:
|
|
"""The number of Repos `search_repos` would find."""
|
|
query = _repo_search_sql(search_str, tag_names)
|
|
if query is None:
|
|
return 0
|
|
from_sql, where, params, _ = query
|
|
return session.execute(text(f'SELECT COUNT(*) FROM {from_sql} {where}'), params).scalar()
|
|
|
|
|
|
async def search_repos_by_name(session: Session, name: str, limit: int = 5) -> List[dict]:
|
|
"""Repos whose name contains `name`, for the search suggestions."""
|
|
if not name:
|
|
return []
|
|
repos = session.query(Repository).options(*Repository.json_options()) \
|
|
.join(Collection, Collection.id == Repository.collection_id) \
|
|
.filter(Collection.name.ilike(f'%{name}%')) \
|
|
.order_by(func.lower(Collection.name)) \
|
|
.limit(limit).all()
|
|
return [dict(id=i.id, name=i.name, tag_name=i.tag_name, location=i.location) for i in repos]
|
|
|
|
|
|
# Fields of one commit, and the end of a commit, in `git log` output.
|
|
LOG_FIELD, LOG_RECORD = '\x1f', '\x1e'
|
|
LOG_FORMAT = LOG_FIELD.join(('%H', '%an', '%cI', '%s')) + LOG_RECORD
|
|
MAXIMUM_LOG_LIMIT = 100
|
|
|
|
|
|
def get_repo_root(repo: Repository) -> pathlib.Path:
|
|
directory = repo.directory
|
|
if not repo.head_sha or not directory or not (directory / '.git').exists():
|
|
raise UnknownDirectory('Repo has not been downloaded')
|
|
return directory
|
|
|
|
|
|
async def _git_output(directory: pathlib.Path, *args) -> Optional[str]:
|
|
"""Run a read-only git command in a Repo; None when it fails."""
|
|
result = await run_command(git_command(*args), cwd=directory, timeout=60, env=git_env(), log_command=False)
|
|
if result.return_code != 0:
|
|
logger.warning(f'git {args[0]} failed in {directory}: {result.stderr.decode(errors="replace")}')
|
|
return None
|
|
return result.stdout.decode(errors='replace')
|
|
|
|
|
|
async def get_repo_log(repo: Repository, limit: int = 50, offset: int = 0) -> dict:
|
|
"""The commits of the checked out branch, newest first: {commits: [{sha, author, date, message}], total}."""
|
|
directory = get_repo_root(repo)
|
|
limit = min(max(int(limit), 1), MAXIMUM_LOG_LIMIT)
|
|
offset = max(int(offset), 0)
|
|
|
|
output = await _git_output(directory, 'log', f'--max-count={limit}', f'--skip={offset}',
|
|
f'--format={LOG_FORMAT}', 'HEAD', '--')
|
|
total = await _git_output(directory, 'rev-list', '--count', 'HEAD', '--')
|
|
commits = []
|
|
for record in (output or '').split(LOG_RECORD):
|
|
record = record.strip('\n')
|
|
if not record:
|
|
continue
|
|
sha, author, date, message = record.split(LOG_FIELD, 3)
|
|
commits.append(dict(sha=sha, author=author, date=date, message=message))
|
|
return dict(commits=commits, total=int(total.strip()) if total else len(commits))
|
|
|
|
|
|
def repo_archive_name(repo: Repository) -> str:
|
|
"""The name of a Repo's ZIP (without `.zip`), safe for a header and as the directory in the ZIP."""
|
|
name = re.sub(r'[^A-Za-z0-9._-]+', '_', repo.name or 'repo').strip('._') or 'repo'
|
|
return f'{name}-{repo.head_sha[:7]}' if repo.head_sha else name
|
|
|
|
|
|
async def stream_repo_archive(directory: pathlib.Path, name: str, send) -> None:
|
|
"""Stream a ZIP of the checked out commit of the repo in `directory` to `send` (an async callable taking
|
|
bytes). The ZIP's files are in the directory `name`.
|
|
|
|
Takes no Repository: the request's session is closed once a streamed response starts. git writes the ZIP as
|
|
it reads the repo, so a large repo is never held in memory."""
|
|
cmd = git_command('archive', '--format=zip', f'--prefix={name}/', 'HEAD')
|
|
proc = await asyncio.create_subprocess_exec(*cmd, cwd=directory, env=git_env(), stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.DEVNULL)
|
|
try:
|
|
while chunk := await proc.stdout.read(64 * 1024):
|
|
await send(chunk)
|
|
await proc.wait()
|
|
finally:
|
|
if proc.returncode is None:
|
|
# The client went away.
|
|
proc.kill()
|
|
await proc.wait()
|
|
|
|
|
|
@dataclass
|
|
class RepoTreeEntry:
|
|
name: str
|
|
path: str
|
|
is_dir: bool
|
|
size: Optional[int] = None
|
|
|
|
def __json__(self) -> dict:
|
|
return dict(name=self.name, path=self.path, is_dir=self.is_dir, size=self.size)
|
|
|
|
|
|
def list_repo_tree(repo: Repository, path: str = '') -> dict:
|
|
"""List one directory of a Repo's checked out files, directories first. `.git` is hidden."""
|
|
root = repo.directory
|
|
if not root or not root.is_dir():
|
|
raise UnknownDirectory('Repo has not been downloaded')
|
|
root = root.resolve()
|
|
|
|
path = (path or '').strip('/')
|
|
directory = (root / path).resolve()
|
|
if directory != root and root not in directory.parents:
|
|
raise InvalidRepo('Path is not in the repo')
|
|
if not directory.is_dir():
|
|
raise UnknownDirectory(f'No directory {path!r} in repo')
|
|
|
|
entries = []
|
|
for entry in directory.iterdir():
|
|
if entry.name == '.git' or entry.is_symlink():
|
|
continue
|
|
relative = str(entry.relative_to(root))
|
|
if entry.is_dir():
|
|
entries.append(RepoTreeEntry(name=entry.name, path=relative, is_dir=True))
|
|
else:
|
|
entries.append(RepoTreeEntry(name=entry.name, path=relative, is_dir=False, size=entry.stat().st_size))
|
|
entries.sort(key=lambda i: (not i.is_dir, i.name.lower()))
|
|
|
|
readme = find_readme(directory)
|
|
return dict(
|
|
path=path,
|
|
entries=[i.__json__() for i in entries],
|
|
readme_path=str((directory / readme).relative_to(root)) if readme else None,
|
|
)
|
|
|
|
|
|
def get_repo_size(directory: pathlib.Path) -> int:
|
|
"""Total size of every file in the repo, including `.git`."""
|
|
total = 0
|
|
for dir_path, _, file_names in os.walk(directory):
|
|
for name in file_names:
|
|
try:
|
|
total += os.lstat(os.path.join(dir_path, name)).st_size
|
|
except FileNotFoundError:
|
|
pass
|
|
return total
|
|
|
|
|
|
@dataclass
|
|
class ReposConfigValidator:
|
|
version: int = 0
|
|
repos: List[dict] = field(default_factory=list)
|
|
|
|
|
|
class ReposConfig(ConfigFile):
|
|
"""Config file for git Repos (Collections with kind='repo').
|
|
|
|
Format:
|
|
repos:
|
|
- name: kiwix-tools
|
|
url: https://github.com/kiwix/kiwix-tools
|
|
directory: repos/software/kiwix-tools
|
|
tag_name: software
|
|
mode: full
|
|
branch: null
|
|
submodules: false
|
|
frequency: 604800
|
|
"""
|
|
file_name = 'repos.yaml'
|
|
validator = ReposConfigValidator
|
|
default_config = dict(version=0, repos=[])
|
|
width = 120
|
|
|
|
@property
|
|
def repos(self) -> List[dict]:
|
|
return self._config.get('repos', [])
|
|
|
|
def import_config(self, file: pathlib.Path = None, send_events=False):
|
|
"""Import Repos from the config into the database."""
|
|
file = file or self.get_file()
|
|
file_str = str(self.get_relative_file())
|
|
|
|
# A missing config is not an empty config; never delete Repos because a file is lost.
|
|
if not file.is_file():
|
|
logger.info(f'No repos config file, skipping import: {file_str}')
|
|
self.successful_import = True
|
|
return
|
|
|
|
super().import_config(file, send_events)
|
|
repos_data = self._config.get('repos', [])
|
|
|
|
# An empty repos list never deletes DB records. Deleting all repos is done through the API.
|
|
if not repos_data:
|
|
logger.info(f'No repos in config, preserving existing DB repos: {file_str}')
|
|
self.successful_import = True
|
|
return
|
|
|
|
try:
|
|
with get_db_session(commit=True) as session:
|
|
imported_ids = set()
|
|
for data in repos_data:
|
|
try:
|
|
repo = self._import_repo(session, data)
|
|
except Exception as e:
|
|
logger.error(f'Failed to import repo from config: {data}', exc_info=e)
|
|
self.import_skipped += 1
|
|
continue
|
|
imported_ids.add(repo.id)
|
|
|
|
for repo in session.query(Repository).all():
|
|
if repo.id not in imported_ids:
|
|
logger.info(f'Deleting repo {repo.name!r} (no longer in config)')
|
|
_delete_repository(session, repo)
|
|
|
|
logger.info(f'Imported {len(imported_ids)} repos from {file_str}')
|
|
self.successful_import = True
|
|
except Exception as e:
|
|
self.successful_import = False
|
|
message = f'Failed to import {file_str} config!'
|
|
logger.error(message, exc_info=e)
|
|
if send_events:
|
|
Events.send_config_import_failed(message)
|
|
raise
|
|
|
|
@staticmethod
|
|
def _import_repo(session: Session, data: dict) -> Repository:
|
|
parsed = parse_repo_url(data['url'])
|
|
name = data.get('name') or parsed.name
|
|
mode = data.get('mode') or 'full'
|
|
_validate_mode(mode)
|
|
frequency = data.get('frequency') or DEFAULT_REPO_FREQUENCY
|
|
_validate_frequency(frequency)
|
|
directory = data.get('directory')
|
|
directory = get_media_directory() / directory if directory else None
|
|
if directory:
|
|
_validate_directory(directory)
|
|
tag_name = data.get('tag_name')
|
|
|
|
repo = get_repository_by_url(session, parsed.url)
|
|
if not repo:
|
|
return create_repository(session, parsed.url, tag_name=tag_name, frequency=frequency, mode=mode,
|
|
branch=data.get('branch'), name=name, description=data.get('description'),
|
|
directory=directory, submodules=bool(data.get('submodules')))
|
|
|
|
collection = repo.collection
|
|
collection.name = name
|
|
repo.update_search_name()
|
|
collection.description = data.get('description')
|
|
collection.tag = Tag.get_or_create_tag(session, tag_name) if tag_name else None
|
|
if directory:
|
|
collection.directory = directory
|
|
repo.mode = mode
|
|
repo.submodules = bool(data.get('submodules'))
|
|
branch = data.get('branch') or None
|
|
_validate_branch(branch)
|
|
repo.branch = branch
|
|
if download := repo.download:
|
|
download.frequency = frequency
|
|
else:
|
|
collection.get_or_create_download(session, repo.url, frequency, downloader_name='git')
|
|
session.flush()
|
|
return repo
|
|
|
|
def dump_config(self, file: pathlib.Path = None, send_events=False, overwrite=False):
|
|
"""Dump all Repos to the config file."""
|
|
with get_db_session() as session:
|
|
repos = session.query(Repository).options(*Repository.json_options()) \
|
|
.join(Collection, Collection.id == Repository.collection_id) \
|
|
.order_by(func.lower(Collection.name)).all()
|
|
data = []
|
|
for repo in repos:
|
|
download = repo.download
|
|
entry = dict(
|
|
name=repo.name,
|
|
url=repo.url,
|
|
directory=str(get_relative_to_media_directory(repo.directory)) if repo.directory else None,
|
|
tag_name=repo.tag_name,
|
|
description=repo.collection.description,
|
|
mode=repo.mode,
|
|
branch=repo.branch,
|
|
submodules=repo.submodules,
|
|
frequency=download.frequency if download else None,
|
|
)
|
|
data.append(entry)
|
|
|
|
self._config['repos'] = data
|
|
logger.info(f'Dumping {len(data)} repos to config')
|
|
self.save(file, send_events, overwrite)
|
|
|
|
def preview_backup_import(self, backup_date: str, mode: str) -> dict:
|
|
"""Preview which Repos a backup import would add/remove (keyed by URL)."""
|
|
backup_data = self.read_config_file(self._get_backup_file(backup_date))
|
|
current_data = self.read_config_file() if self.get_file().is_file() else dict(repos=[], version=0)
|
|
|
|
backup_repos = {repo_url_key(i['url']): i for i in backup_data.get('repos', []) if i.get('url')}
|
|
current_repos = {repo_url_key(i['url']): i for i in current_data.get('repos', []) if i.get('url')}
|
|
|
|
add = [dict(type='repo', name=i.get('name', ''), url=i['url'])
|
|
for key, i in backup_repos.items() if key not in current_repos]
|
|
unchanged = len([i for i in backup_repos if i in current_repos])
|
|
remove = []
|
|
if mode == 'overwrite':
|
|
remove = [dict(type='repo', name=i.get('name', ''), url=i['url'])
|
|
for key, i in sorted(current_repos.items()) if key not in backup_repos]
|
|
return dict(mode=mode, add=add, remove=remove, unchanged=unchanged)
|
|
|
|
def import_backup(self, backup_date: str, mode: str, send_events: bool = False):
|
|
"""Restore Repos from a dated backup of repos.yaml ('overwrite' or 'merge' by URL)."""
|
|
self._preserve_current_config()
|
|
backup_file = self._get_backup_file(backup_date)
|
|
config_file = self.get_file()
|
|
config_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if mode == 'overwrite':
|
|
shutil.copy2(backup_file, config_file)
|
|
elif mode == 'merge':
|
|
backup_data = self.read_config_file(backup_file)
|
|
current_data = self.read_config_file() if config_file.is_file() else dict(repos=[], version=0)
|
|
current_keys = {repo_url_key(i['url']) for i in current_data.get('repos', []) if i.get('url')}
|
|
merged = list(current_data.get('repos', []))
|
|
merged.extend(i for i in backup_data.get('repos', [])
|
|
if i.get('url') and repo_url_key(i['url']) not in current_keys)
|
|
self.write_config_data(dict(repos=merged, version=current_data.get('version', 0) + 1), config_file)
|
|
|
|
self.import_config(send_events=send_events)
|
|
save_downloads_config.activate_switch()
|
|
|
|
|
|
repos_config = ReposConfig()
|
|
|
|
|
|
def get_repos_config() -> ReposConfig:
|
|
return repos_config
|
|
|
|
|
|
@register_switch_handler('save_repos_config')
|
|
def save_repos_config():
|
|
"""Save the repos config when the switch is activated."""
|
|
repos_config.background_dump.activate_switch()
|
|
|
|
|
|
save_repos_config: ActivateSwitchMethod
|
|
|
|
|
|
def import_repos_config():
|
|
logger.info('Importing repos config')
|
|
repos_config.import_config()
|