#!/usr/bin/env python3 """ Интеграция GitFlic → PT Sandbox ================================= PoC-скрипт для автоматической выгрузки релизов из GitFlic в PT Sandbox на анализ. """ import json import logging import sys import ssl import time import warnings from copy import deepcopy from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import requests import yaml from requests.adapters import HTTPAdapter from urllib3.poolmanager import PoolManager class InsecureSslAdapter(HTTPAdapter): """Adapter that bypasses SSL verification entirely.""" def init_poolmanager(self, *args, **kwargs): ctx = ssl._create_unverified_context() kwargs["ssl_context"] = ctx return super().init_poolmanager(*args, **kwargs) def proxy_manager_for(self, *args, **kwargs): ctx = ssl._create_unverified_context() kwargs["ssl_context"] = ctx return super().proxy_manager_for(*args, **kwargs) logger = logging.getLogger("sandbox_gitflic") def setup_logging(cfg: dict) -> None: """Настройка логирования из конфигурации.""" log_cfg = cfg.get("logging", {}) level_name = log_cfg.get("level", "INFO").upper() level = getattr(logging, level_name, logging.INFO) fmt = log_cfg.get("format", "%(asctime)s [%(levelname)s] %(name)s: %(message)s") handlers: List[logging.Handler] = [logging.StreamHandler(sys.stdout)] log_file = log_cfg.get("file") if log_file: handlers.append(logging.FileHandler(log_file, encoding="utf-8")) logging.basicConfig(level=level, format=fmt, handlers=handlers) logger.setLevel(level) def load_config(path: str = "config.yaml") -> dict: """Загрузить YAML-конфигурацию.""" path = Path(path) if not path.exists(): logger.error("Файл конфигурации не найден: %s", path.absolute()) sys.exit(1) with open(path, "r", encoding="utf-8") as fh: cfg = yaml.safe_load(fh) logger.info("Конфигурация загружена из %s", path) return cfg # ───────────────────────────────────────────────────────── # Управление состоянием # ───────────────────────────────────────────────────────── def load_state(path: str) -> dict: """ Загрузить состояние из JSON-файла. Формат: { "sent_releases": { ":::": { "project_id": int, "project_name": str, "release_id": int, "release_tag": str, "release_version": str, "files": [{"filename": str, "sandbox_task_id": str|null}], "sent_at": "2024-01-01T00:00:00Z" } } } """ path_obj = Path(path) if not path_obj.exists(): logger.info("Файл состояния не найден. Начинаем с пустого.") return {"sent_releases": {}} try: with open(path_obj, "r", encoding="utf-8") as fh: state = json.load(fh) logger.info("Состояние загружено из %s (%d релизов отправлено)", path, len(state.get("sent_releases", {}))) return state except (json.JSONDecodeError, OSError) as exc: logger.error("Ошибка загрузки состояния из %s: %s. Начинаем с пустого.", path, exc) return {"sent_releases": {}} def save_state(state: dict, path: str) -> None: """Сохранить состояние в JSON-файл (атомарно через rename).""" path_obj = Path(path) tmp_path = path_obj.with_suffix(".json.tmp") try: with open(tmp_path, "w", encoding="utf-8") as fh: json.dump(state, fh, ensure_ascii=False, indent=2) tmp_path.replace(path_obj) logger.debug("Состояние сохранено в %s", path) except OSError as exc: logger.error("Ошибка сохранения состояния в %s: %s", path, exc) def make_release_key(project_id: str, release_id: str, tag: str, version: str) -> str: """Сформировать ключ для идентификации релиза.""" return f"{project_id}:{release_id}:{tag}:{version}" def is_release_sent(state: dict, project_id: str, release_id: str, tag: str, version: str) -> bool: """Проверить, был ли релиз уже отправлен.""" key = make_release_key(project_id, release_id, tag, version) return key in state.get("sent_releases", {}) def mark_release_sent(state: dict, project_id: str, project_name: str, release_id: str, tag: str, version: str, files: List[Dict[str, Any]]) -> None: """Пометить релиз как отправленный.""" key = make_release_key(project_id, release_id, tag, version) if "sent_releases" not in state: state["sent_releases"] = {} state["sent_releases"][key] = { "project_id": project_id, "project_name": project_name, "release_id": release_id, "release_tag": tag, "release_version": version, "files": files, "sent_at": datetime.now(timezone.utc).isoformat(), } logger.info("Релиз помечен как отправленный: %s", key) # ───────────────────────────────────────────────────────── # API GitFlic # ───────────────────────────────────────────────────────── class GitFlicClient: """Клиент для REST API GitFlic.""" def __init__(self, cfg: dict): gf_cfg = cfg["gitflic"] self.base_url = gf_cfg["base_url"].rstrip("/") self.api_token = gf_cfg["api_token"] self.per_page = gf_cfg.get("per_page", 50) self.chunk_size = gf_cfg.get("download_chunk_size", 8 * 1024 * 1024) self.verify_ssl = gf_cfg.get("verify_ssl", True) if not self.api_token: logger.error("API-токен GitFlic не задан в конфигурации!") sys.exit(1) self._session = requests.Session() self._session.headers.update({ "Authorization": f"token {self.api_token}", "Accept": "application/json", "User-Agent": "Sandbox-GitFlic-PoC/1.0", }) if not self.verify_ssl: self._session.verify = False self._session.mount("https://", InsecureSslAdapter()) logger.warning("Проверка SSL-сертификата отключена для GitFlic") def _request(self, method: str, path: str, **kwargs) -> requests.Response: """Выполнить HTTP-запрос к API GitFlic.""" url = f"{self.base_url}{path}" kwargs.setdefault("verify", self.verify_ssl) try: resp = self._session.request(method, url, **kwargs) resp.raise_for_status() return resp except requests.exceptions.Timeout: logger.error("Тайм-аут при запросе %s %s", method, url) raise except requests.exceptions.HTTPError as exc: logger.error("HTTP-ошибка %s: %s - %s", resp.status_code, resp.reason, resp.text[:500]) raise except requests.exceptions.RequestException as exc: logger.error("Ошибка запроса %s %s: %s", method, url, exc) raise def _paginated_get(self, path: str, params: Optional[dict] = None, list_key: Optional[str] = None) -> List[dict]: """ Получить все страницы пагинированного GET-запроса. GitFlic использует: size (не per_page), page начиная с 0, и ответ в формате {"_embedded": {"": [...]}, "page": {...}}. """ if params is None: params = {} params.setdefault("size", self.per_page) page = 0 all_items: List[dict] = [] while True: params["page"] = page resp = self._request("GET", path, params=params) data = resp.json() if list_key: items = data.get("_embedded", {}).get(list_key, []) else: embedded = data.get("_embedded", {}) items = next(iter(embedded.values())) if embedded else [] all_items.extend(items) page_info = data.get("page", {}) total_pages = page_info.get("totalPages", 0) if page >= total_pages - 1: break page += 1 return all_items def get_my_projects(self) -> List[dict]: """GET /project/my — собственные проекты пользователя.""" return self._paginated_get("/project/my", list_key="projectList") def get_shared_projects(self) -> List[dict]: """GET /project/shared — проекты с участием пользователя.""" return self._paginated_get("/project/shared", list_key="projectList") def resolve_projects(self, cfg: dict) -> List[dict]: """Определить список проектов для сканирования.""" gf_cfg = cfg["gitflic"] source = gf_cfg.get("project_source", "all") if source == "list": names = gf_cfg.get("project_list", []) ids = gf_cfg.get("project_ids", []) projects = [] for name in names: proj = self.get_project_by_path(name) if proj: projects.append(proj) else: logger.warning("Проект не найден: %s", name) for pid in ids: proj = self.get_project(pid) if proj: projects.append(proj) else: logger.warning("Проект не найден по ID: %s", pid) return projects if source == "company": cid = gf_cfg.get("company_id") if not cid: logger.error("Не указан company_id при project_source=company") return [] return self.get_company_projects(cid) if source == "team": tid = gf_cfg.get("team_id") if not tid: logger.error("Не указан team_id при project_source=team") return [] return self.get_team_projects(tid) return self.get_shared_projects() def get_project(self, project_id: str) -> Optional[dict]: """GET /project/{projectUuid}.""" try: resp = self._request("GET", f"/project/{project_id}") return resp.json() except requests.RequestException: return None def get_project_by_path(self, path: str) -> Optional[dict]: """ GET /project/{ownerAlias}/{projectAlias}. path — строка вида "owner/project-name". """ try: resp = self._request("GET", f"/project/{path}") return resp.json() except requests.RequestException: return None def get_releases(self, owner_alias: str, project_alias: str) -> List[dict]: """GET /project/{ownerAlias}/{projectAlias}/release.""" return self._paginated_get( f"/project/{owner_alias}/{project_alias}/release", list_key="releaseTagModelList", ) def get_company_projects(self, company_id: str) -> List[dict]: """GET /company/{id}/project.""" return self._paginated_get( f"/company/{company_id}/project", list_key="projectList", ) def get_team_projects(self, team_id: str) -> List[dict]: """GET /team/{id}/project.""" return self._paginated_get( f"/team/{team_id}/project", list_key="projectList", ) def download_file(self, url: str, local_path: Path) -> Path: """Скачать файл по ссылке.""" if not url.startswith("http"): url = f"{self.base_url}{url}" headers = {} if url.startswith(self.base_url): headers["Authorization"] = f"token {self.api_token}" logger.info("Скачивание: %s -> %s", url, local_path) resp = self._session.get(url, headers=headers, stream=True, timeout=(30, 300), verify=self.verify_ssl) resp.raise_for_status() local_path.parent.mkdir(parents=True, exist_ok=True) with open(local_path, "wb") as fh: for chunk in resp.iter_content(chunk_size=self.chunk_size): if chunk: fh.write(chunk) logger.debug("Скачан файл: %s (%d bytes)", local_path, local_path.stat().st_size) return local_path # ───────────────────────────────────────────────────────── # Клиент PT Sandbox # ───────────────────────────────────────────────────────── class PTSandboxClient: """Клиент для REST API PT Sandbox.""" def __init__(self, cfg: dict): sb_cfg = cfg["pt_sandbox"] self.base_url = sb_cfg["base_url"].rstrip("/") self.upload_endpoint = sb_cfg.get("upload_endpoint", "/api/v1/scan/checkFile") self.auth_header = sb_cfg.get("auth_header", "X-API-Key") self.api_key = sb_cfg.get("api_key", "") self.file_field_name = sb_cfg.get("file_field_name", "file") self.extra_fields = sb_cfg.get("extra_fields", {}) self.timeout_connect = sb_cfg.get("timeout_connect", 30) self.timeout_read = sb_cfg.get("timeout_read", 300) self.verify_ssl = sb_cfg.get("verify_ssl", True) if not self.api_key: logger.error("API-ключ PT Sandbox не задан в конфигурации!") sys.exit(1) self._session = requests.Session() self._session.headers.update({ self.auth_header: self.api_key, "User-Agent": "Sandbox-GitFlic-PoC/1.0", }) if not self.verify_ssl: self._session.verify = False self._session.mount("https://", InsecureSslAdapter()) logger.warning("Проверка SSL-сертификата отключена для PT Sandbox") def upload_file(self, file_path: Path) -> Optional[str]: """ Отправить файл в PT Sandbox на анализ (одним запросом — загрузка + запуск проверки). Возвращает ID задачи (scan_id) или None при ошибке. """ params = { "file_name": file_path.name, "async_result": "true", } url = f"{self.base_url}{self.upload_endpoint}" logger.info("Отправка файла в Sandbox: %s (%d bytes)", file_path.name, file_path.stat().st_size) try: with open(file_path, "rb") as fh: files = {self.file_field_name: (file_path.name, fh)} fields = dict(self.extra_fields) resp = self._session.post( url, params=params, files=files, data=fields, timeout=(self.timeout_connect, self.timeout_read), verify=self.verify_ssl, ) if resp.status_code in (200, 201, 202): result = resp.json() data = result.get("data", {}) scan_id = data.get("scan_id") or result.get("task_id") or result.get("id") logger.info("Файл %s принят Sandbox. scan_id=%s", file_path.name, scan_id) return str(scan_id) if scan_id else None else: logger.error("Sandbox вернул %s: %s", resp.status_code, resp.text[:500]) return None except requests.exceptions.Timeout: logger.error("Тайм-аут при отправке файла %s в Sandbox", file_path.name) return None except requests.exceptions.RequestException as exc: logger.error("Ошибка отправки файла %s в Sandbox: %s", file_path.name, exc) return None except OSError as exc: logger.error("Ошибка чтения файла %s: %s", file_path.name, exc) return None # ───────────────────────────────────────────────────────── # Вспомогательные функции # ───────────────────────────────────────────────────────── def get_temp_dir() -> Path: """Получить временную директорию для скачанных файлов.""" tmp = Path("downloads") tmp.mkdir(parents=True, exist_ok=True) return tmp # ───────────────────────────────────────────────────────── # Основная логика сканирования # ───────────────────────────────────────────────────────── def scan_new_releases(gf: GitFlicClient, sb: PTSandboxClient, state: dict, cfg: dict) -> dict: """ Основная функция: 1. Получить проекты 2. Для каждого проекта релизы 3. Отфильтровать отправленные 4. Скачать файлы (attachmentFiles из релиза) 5. Отправить в Sandbox 6. Обновить состояние """ projects = gf.resolve_projects(cfg) if not projects: logger.warning("Нет проектов для сканирования.") return state scan_mode = cfg["gitflic"].get("scan_mode", "latest") tmp_dir = get_temp_dir() state = deepcopy(state) for project in projects: pid = project.get("id") owner_alias = project.get("owner", {}).get("alias", "") project_alias = project.get("alias", "") pname = f"{owner_alias}/{project_alias}" logger.info("Обработка проекта: %s (uuid=%s)", pname, pid) if not owner_alias or not project_alias: logger.warning("Пропускаем проект %s: нет owner/project alias", pid) continue try: releases = gf.get_releases(owner_alias, project_alias) except requests.RequestException: logger.error("Пропускаем проект %s - ошибка получения релизов", pname) continue if not releases: logger.info("Нет релизов в проекте %s", pname) continue releases.sort(key=lambda r: r.get("createdAt", ""), reverse=True) if scan_mode == "latest": rel = releases[0] rel_id = rel.get("id") rel_tag = rel.get("tagName", "") rel_version = rel.get("title", "") if is_release_sent(state, pid, rel_id, rel_tag, rel_version): logger.info("Последний релиз уже отправлен. Пропускаем.") continue releases_to_process = [rel] else: releases_to_process = [ r for r in releases if not is_release_sent(state, pid, r.get("id"), r.get("tagName", ""), r.get("title", "")) ] if not releases_to_process: logger.info("Нет новых релизов в проекте %s", pname) continue for rel in releases_to_process: rel_id = rel.get("id") rel_tag = rel.get("tagName", "") rel_version = rel.get("title", "") rel_key = make_release_key(pid, rel_id, rel_tag, rel_version) logger.info("Новый релиз: %s (tag=%s, title=%s)", rel_key, rel_tag, rel_version) # Собираем файлы для скачивания (attachmentFiles из релиза) files_to_download: List[Dict[str, Any]] = [] local_files: List[Dict[str, Any]] = [] assets = rel.get("attachmentFiles", []) for asset in assets: aname = asset.get("name", "") aurl = asset.get("link", "") if not aname or not aurl: continue files_to_download.append({"type": "asset", "url": aurl, "filename": aname}) if not files_to_download: logger.info("Нет файлов для скачивания в релизе %s", rel_key) # Скачиваем все файлы for fi in files_to_download: safe = fi["filename"].replace("/", "_").replace("\\", "_") lpath = tmp_dir / f"{pid}_{rel_id}_{safe}" try: gf.download_file(fi["url"], lpath) local_files.append({"filename": fi["filename"], "type": fi["type"], "local_path": str(lpath)}) except requests.RequestException as exc: logger.error("Ошибка скачивания %s: %s", fi["filename"], exc) # Отправляем каждый файл в Sandbox sent_files: List[Dict[str, Any]] = [] for lf in local_files: lpath = Path(lf["local_path"]) if not lpath.exists(): logger.warning("Файл %s не найден", lpath) continue task_id = sb.upload_file(lpath) sent_files.append({"filename": lf["filename"], "type": lf["type"], "sandbox_task_id": task_id}) try: lpath.unlink() except OSError as exc: logger.warning("Не удалось удалить %s: %s", lpath, exc) mark_release_sent(state, pid, pname, rel_id, rel_tag, rel_version, sent_files) logger.info("Релиз %s обработан: %d файлов отправлено", rel_key, len(sent_files)) return state # ───────────────────────────────────────────────────────── # Точка входа # ───────────────────────────────────────────────────────── def main() -> None: """Главная функция.""" cfg = load_config() setup_logging(cfg) logger.info("=" * 60) logger.info("Запуск интеграции GitFlic -> PT Sandbox") logger.info("=" * 60) gf = GitFlicClient(cfg) sb = PTSandboxClient(cfg) loop_cfg = cfg.get("loop", {}) loop_enabled = loop_cfg.get("enabled", False) interval = loop_cfg.get("interval_seconds", 3600) state_path = cfg.get("state", {}).get("file_path", "state.json") iteration = 0 while True: iteration += 1 logger.info("Итерация #%d", iteration) state = load_state(state_path) try: new_state = scan_new_releases(gf, sb, state, cfg) save_state(new_state, state_path) except Exception as exc: logger.exception("Критическая ошибка на итерации #%d: %s", iteration, exc) try: save_state(state, state_path) except Exception: pass if not loop_enabled: logger.info("Однократный проход завершён.") break logger.info("Ожидание %d секунд до следующей итерации...", interval) try: time.sleep(interval) except KeyboardInterrupt: logger.info("Получен сигнал прерывания. Завершение.") break logger.info("Работа завершена.") if __name__ == "__main__": main()