"""The only component allowed to contact Gitea. Never logs payloads or credentials.""" from dataclasses import dataclass, field import logging import os import re from urllib.parse import quote, urlsplit import httpx logger = logging.getLogger(__name__) class GiteaError(Exception): def __init__(self, kind, uncertain=False): self.kind = kind self.uncertain = uncertain super().__init__(kind) def redact(text, secrets=()): for secret in secrets: if secret and len(secret) >= 8: text = text.replace(secret, '[REDACTED]') text = re.sub(r'-----BEGIN [^-]*PRIVATE KEY-----.*?-----END [^-]*PRIVATE KEY-----', '[REDACTED PRIVATE KEY]', text, flags=re.S) text = re.sub(r'(?im)((?:password|passwd|authorization|cookie|[\w]*(?:token|secret|credential|api_key)[\w]*)\s*[:=]\s*)[^\r\n]+', r'\1[REDACTED]', text) text = re.sub(r'[A-Za-z0-9_-]{20,}:[A-Za-z0-9_-]{80,}', '[REDACTED FCM TOKEN]', text) text = re.sub(r'(https?://)[^/\s:@]+:[^/\s@]+@', r'\1[REDACTED]@', text) return text.replace('\x00', '') @dataclass(frozen=True) class GiteaConfig: url: str token: str = field(repr=False) owner: str repo: str @classmethod def from_env(cls): return cls(*(os.environ.get(key, '').strip() for key in ('GITEA_URL', 'GITEA_TOKEN', 'GITEA_OWNER', 'GITEA_REPO'))) def validate(self): try: parsed = urlsplit(self.url) except ValueError: raise GiteaError('configuration') from None if (not all((self.url, self.token, self.owner, self.repo)) or parsed.scheme not in ('http', 'https') or not parsed.netloc or parsed.username or parsed.password or parsed.query or parsed.fragment or not re.fullmatch(r'[A-Za-z0-9_.-]+', self.owner) or not re.fullmatch(r'[A-Za-z0-9_.-]+', self.repo)): raise GiteaError('configuration') class GiteaService: def __init__(self, config=None, transport=None): self.config = config or GiteaConfig.from_env() self.transport = transport def create_issue(self, title, body, category, screenshot=None): self.config.validate() repository = '/repos/' + quote(self.config.owner, safe='') + '/' + quote(self.config.repo, safe='') wanted = {'reported-from-metalcircle', 'bug'} wanted.update({'android': {'android'}, 'web': {'web', 'frontend'}, 'push': {'push'}}.get(category, set())) attempted = False issue_created = False try: with httpx.Client(base_url=self.config.url.rstrip('/') + '/api/v1/', headers={'Authorization': 'token ' + self.config.token, 'Accept': 'application/json'}, timeout=httpx.Timeout(15.0, connect=5.0), follow_redirects=False, transport=self.transport, trust_env=False) as client: # Fail closed if a personal account token is accidentally configured. identity = client.get('user') identity.raise_for_status() if identity.json().get('login') != 'metalcircle-bot': raise GiteaError('wrong_account') labels = [] try: for page in range(1, 11): result = client.get(repository.lstrip('/') + '/labels', params={'limit': 50, 'page': page}) result.raise_for_status() items = result.json() if not isinstance(items, list): break labels.extend(item['id'] for item in items if isinstance(item, dict) and isinstance(item.get('name'), str) and item.get('name', '').lower() in wanted and type(item.get('id')) is int) if len(items) < 50: break except (httpx.HTTPError, ValueError, TypeError, KeyError): # Labels are optional; permissions/missing labels must not block a report. labels = [] attempted = True result = client.post(repository.lstrip('/') + '/issues', json={ 'title': redact('[MetalCircle Bug] ' + title, (self.config.token,)), 'body': redact(body, (self.config.token,)), 'labels': sorted(set(labels)), }) result.raise_for_status() issue = result.json() if (not isinstance(issue, dict) or type(issue.get('number')) is not int or issue['number'] <= 0 or not isinstance(issue.get('user'), dict) or issue['user'].get('login') != 'metalcircle-bot'): raise GiteaError('invalid_response', uncertain=True) issue_number = issue['number'] issue_created = True if screenshot: filename, image_bytes, media_type = screenshot uploaded = client.post( repository.lstrip('/') + f'/issues/{issue_number}/assets', params={'name': filename}, files={'attachment': (filename, image_bytes, media_type)}, ) uploaded.raise_for_status() attachment = uploaded.json() download_url = attachment.get('browser_download_url') if isinstance(attachment, dict) else None configured_origin = urlsplit(self.config.url) attachment_origin = urlsplit(download_url) if isinstance(download_url, str) else None if (not attachment_origin or attachment_origin.scheme not in ('http', 'https') or attachment_origin.hostname != configured_origin.hostname or attachment_origin.port != configured_origin.port or attachment_origin.username or attachment_origin.password): raise GiteaError('invalid_attachment_response', uncertain=True) linked_body = redact( body + '\n\n## Screenshot\n\n![Bug report screenshot](<' + quote(download_url, safe=':/?&=%') + '>)', (self.config.token,), ) updated = client.patch(repository.lstrip('/') + f'/issues/{issue_number}', json={'body': linked_body}) updated.raise_for_status() return issue_number except GiteaError as error: if issue_created and screenshot: error.uncertain = True logger.warning('Gitea issue submission failed: %s', error.kind) raise except httpx.HTTPStatusError as error: code = error.response.status_code logger.warning('Gitea issue submission failed: HTTP %d', code) raise GiteaError('http_' + str(code), uncertain=(issue_created and bool(screenshot)) or (attempted and code >= 500)) from None except (httpx.TimeoutException, httpx.NetworkError, httpx.RemoteProtocolError): logger.warning('Gitea issue submission failed: network_or_timeout') raise GiteaError('network_or_timeout', uncertain=issue_created or attempted) from None except (ValueError, TypeError, KeyError, AttributeError, httpx.HTTPError): logger.warning('Gitea issue submission failed: invalid_response') raise GiteaError('invalid_response', uncertain=issue_created or attempted) from None