122 lines
5.7 KiB
Python
122 lines
5.7 KiB
Python
"""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):
|
|
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
|
|
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)
|
|
return issue['number']
|
|
except GiteaError as error:
|
|
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=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=attempted) from None
|
|
except (ValueError, TypeError, KeyError, AttributeError, httpx.HTTPError):
|
|
logger.warning('Gitea issue submission failed: invalid_response')
|
|
raise GiteaError('invalid_response', uncertain=attempted) from None
|