125 lines
5.0 KiB
Python
125 lines
5.0 KiB
Python
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
|
|
|
|
from app.schemas.swos_beta import SwosBetaCredentials, SwosBetaProbeResponse
|
|
|
|
|
|
@dataclass
|
|
class DownloadedSwosBackup:
|
|
filename: str
|
|
content: bytes
|
|
content_type: str
|
|
auth_mode: str
|
|
base_url: str
|
|
|
|
|
|
class SwosBetaService:
|
|
timeout_seconds = 12
|
|
|
|
def probe(self, payload: SwosBetaCredentials) -> SwosBetaProbeResponse:
|
|
base_url = self._build_base_url(payload.host, payload.port)
|
|
response, auth_mode = self._request_with_fallback('GET', base_url, payload)
|
|
html = response.text if 'text' in (response.headers.get('content-type') or '').lower() else ''
|
|
title = self._extract_title(html)
|
|
|
|
backup_response, _ = self._request_with_fallback('GET', f'{base_url}/backup.swb', payload, allow_text_fallback=False)
|
|
backup_ok = backup_response.status_code == 200 and len(backup_response.content) > 0
|
|
|
|
return SwosBetaProbeResponse(
|
|
success=response.ok,
|
|
base_url=base_url,
|
|
status_code=response.status_code,
|
|
auth_mode=auth_mode,
|
|
page_title=title,
|
|
content_type=response.headers.get('content-type'),
|
|
server=response.headers.get('server'),
|
|
save_backup_visible='save backup' in html.lower(),
|
|
backup_endpoint_ok=backup_ok,
|
|
note='Moduł działa osobno i nie zapisuje kopii do głównego repozytorium.'
|
|
)
|
|
|
|
def download_backup(self, payload: SwosBetaCredentials) -> DownloadedSwosBackup:
|
|
base_url = self._build_base_url(payload.host, payload.port)
|
|
response, auth_mode = self._request_with_fallback('GET', f'{base_url}/backup.swb', payload, allow_text_fallback=False)
|
|
if response.status_code != 200:
|
|
raise ValueError(f'Urządzenie zwróciło kod HTTP {response.status_code} dla /backup.swb.')
|
|
if not response.content:
|
|
raise ValueError('Urządzenie zwróciło pusty plik backupu.')
|
|
|
|
filename = self._build_filename(payload)
|
|
content_type = response.headers.get('content-type') or 'application/octet-stream'
|
|
return DownloadedSwosBackup(
|
|
filename=filename,
|
|
content=response.content,
|
|
content_type=content_type,
|
|
auth_mode=auth_mode,
|
|
base_url=base_url,
|
|
)
|
|
|
|
def _request_with_fallback(self, method: str, url: str, payload: SwosBetaCredentials, allow_text_fallback: bool = True):
|
|
attempts = []
|
|
auth_variants = [
|
|
('digest', HTTPDigestAuth(payload.username, payload.password)),
|
|
('basic', HTTPBasicAuth(payload.username, payload.password)),
|
|
]
|
|
if allow_text_fallback:
|
|
auth_variants.append(('none', None))
|
|
|
|
last_response = None
|
|
for label, auth in auth_variants:
|
|
try:
|
|
response = requests.request(
|
|
method,
|
|
url,
|
|
auth=auth,
|
|
timeout=self.timeout_seconds,
|
|
allow_redirects=True,
|
|
)
|
|
last_response = response
|
|
if response.status_code < 400:
|
|
return response, label
|
|
attempts.append(f'{label}:{response.status_code}')
|
|
except requests.RequestException as exc:
|
|
attempts.append(f'{label}:{exc.__class__.__name__}')
|
|
|
|
if last_response is not None:
|
|
raise ValueError(f'Nie udało się połączyć ze SwOS ({", ".join(attempts)}).')
|
|
raise ValueError('Nie udało się połączyć ze SwOS.')
|
|
|
|
def _build_base_url(self, host: str, port: int) -> str:
|
|
raw = host.strip()
|
|
parsed = urlparse(raw if '://' in raw else f'http://{raw}')
|
|
scheme = parsed.scheme or 'http'
|
|
if scheme not in {'http', 'https'}:
|
|
raise ValueError('Dozwolone są tylko adresy HTTP lub HTTPS.')
|
|
if not parsed.hostname:
|
|
raise ValueError('Nieprawidłowy adres hosta.')
|
|
resolved_port = parsed.port or port
|
|
base = f'{scheme}://{parsed.hostname}'
|
|
if resolved_port not in {80, 443} or (scheme == 'http' and resolved_port != 80) or (scheme == 'https' and resolved_port != 443):
|
|
base = f'{base}:{resolved_port}'
|
|
return base.rstrip('/')
|
|
|
|
def _extract_title(self, html: str) -> str | None:
|
|
if not html:
|
|
return None
|
|
match = re.search(r'<title>(.*?)</title>', html, flags=re.IGNORECASE | re.DOTALL)
|
|
if not match:
|
|
return None
|
|
return re.sub(r'\s+', ' ', match.group(1)).strip() or None
|
|
|
|
def _build_filename(self, payload: SwosBetaCredentials) -> str:
|
|
label = payload.label or payload.host
|
|
safe = re.sub(r'[^A-Za-z0-9._-]+', '-', label).strip('-') or 'switchos'
|
|
timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
|
|
return f'{safe}-swos-{timestamp}.swb'
|
|
|
|
|
|
swos_beta_service = SwosBetaService()
|