Files
sandbox-gitflic/main.py
T

655 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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>:<release_id>:<tag>:<version>": {
"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": {"<listKey>": [...]}, "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/storage/uploadScanFile")
self.scan_endpoint = sb_cfg.get("scan_endpoint", "/api/v1/analysis/createScanTask")
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.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 _request(self, method: str, url: str, **kwargs) -> requests.Response:
"""Выполнить HTTP-запрос к API PT Sandbox."""
try:
resp = self._session.request(method, url, **kwargs)
if resp.status_code == 401:
logger.error("Sandbox 401: неверный или неактивный токен доступа")
return resp
resp.raise_for_status()
return resp
except requests.exceptions.RequestException as exc:
logger.error("Ошибка запроса Sandbox %s: %s", url, exc)
raise
def upload_file(self, file_path: Path) -> Optional[str]:
"""
Двухшаговый流程: загрузить файл в Sandbox, затем запустить проверку.
Возвращает scan_id или None при ошибке.
"""
# Шаг 1: загрузка файла
upload_url = f"{self.base_url}{self.upload_endpoint}"
logger.info("Шаг 1/2: загрузка файла %s в Sandbox", file_path.name)
try:
with open(file_path, "rb") as fh:
files = {self.file_field_name: (file_path.name, fh)}
resp = self._session.post(
upload_url, files=files,
timeout=(self.timeout_connect, self.timeout_read),
verify=self.verify_ssl,
)
if resp.status_code not in (200, 201, 202):
logger.error("Sandbox ошибка загрузки: %s: %s",
resp.status_code, resp.text[:500])
return None
upload_data = resp.json().get("data", {})
file_uri = upload_data.get("file_uri")
if not file_uri:
logger.error("Sandbox: не получен file_uri в ответе на загрузку")
return None
logger.debug("Файл загружен, file_uri=%s", file_uri)
except requests.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
# Шаг 2: запуск проверки
scan_url = f"{self.base_url}{self.scan_endpoint}"
logger.info("Шаг 2/2: запуск проверки файла %s", file_path.name)
try:
body = {
"file_uri": file_uri,
"file_name": file_path.name,
"async_result": True,
"short_result": True,
}
resp = self._session.post(
scan_url, json=body,
timeout=(self.timeout_connect, self.timeout_read),
verify=self.verify_ssl,
)
if resp.status_code not in (200, 201, 202):
logger.error("Sandbox ошибка запуска проверки: %s: %s",
resp.status_code, resp.text[:500])
return None
scan_data = resp.json().get("data", {})
scan_id = scan_data.get("scan_id") or scan_data.get("id")
if scan_id:
logger.info("Файл %s принят Sandbox. scan_id=%s",
file_path.name, scan_id)
return str(scan_id)
logger.error("Sandbox: не получен scan_id в ответе")
return None
except requests.RequestException as exc:
logger.error("Ошибка запуска проверки %s в Sandbox: %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()