mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
48a7024b96
While auditing real-world Bambuddy backup repos on GitHub I found
several left public. That's a serious leak: the settings backup only
filters bambu_cloud_token and auth_secret_key, so mqtt_username,
mqtt_password, ha_token, prometheus_token, bambu_cloud_email,
external_url, and the printer access codes (via K-profiles) were going
to whatever visibility the user picked.
Hard guard at every save and re-checked on every push:
- POST /github-backup/config and PATCH /github-backup/config (when URL,
token, or provider changes) run a connection test internally and
return 400 unless is_private comes back True.
- run_backup() re-checks before each scheduled or manual push, so a
repository that flipped from private to public gets a clear
"Backup aborted: the target repository is no longer private" failure.
Each provider's test_connection now returns is_private (GitHub /
Gitea / Forgejo read data.private, GitLab reads visibility=="private";
"internal" is treated as non-private). None means "couldn't determine"
and is also rejected -- safer to fail closed.
Frontend renders visibility inline on Test Connection: green check when
private, red warning panel listing every credential at risk when public,
yellow when unknown.
---
ui(github-backup): show save-failure messages inline on the card
The new "repository is not private" rejection message is ~250 characters
listing every credential the backup carries (MQTT password, HA token,
Prometheus token, Bambu Cloud email, printer access codes), which clips
badly in a toast.
Both the initial-setup save and the debounced autosave now stash the
backend's error message into a saveError state and render it as a red
inline banner above the test-result block, with whitespace-pre-wrap so
the full message stays readable. The banner clears on success, on the
next save attempt, and when the user starts editing URL / token / provider
-- the three fields whose changes invalidate the privacy check -- so it
doesn't linger after the user has already addressed the cause.
Short success toasts (Settings saved, Token updated, Backup enabled) are
unchanged.
267 lines
11 KiB
Python
267 lines
11 KiB
Python
"""GitLab backend — implements GitProviderBackend using the GitLab REST API v4."""
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import re
|
|
import urllib.parse
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
|
|
from backend.app.services.git_providers.base import GitProviderBackend
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GitLabBackend(GitProviderBackend):
|
|
"""Backend for gitlab.com and self-hosted GitLab instances."""
|
|
|
|
def get_api_base(self, repo_url: str) -> str:
|
|
match = re.match(r"(https?://[\w.\-]+(:\d+)?)/", repo_url)
|
|
if not match:
|
|
raise ValueError(f"Cannot derive API base from URL: {repo_url}")
|
|
return f"{match.group(1)}/api/v4"
|
|
|
|
def get_headers(self, token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
|
|
|
def parse_repo_url(self, url: str) -> tuple[str, str]:
|
|
"""Return (namespace, repo) from HTTPS or SSH URL.
|
|
|
|
namespace may include subgroups, e.g. 'group/subgroup' for
|
|
gitlab.com/group/subgroup/project. Callers join them with '/' and
|
|
URL-encode the result for /api/v4/projects/{encoded_path}.
|
|
"""
|
|
if not url or len(url) > 500:
|
|
raise ValueError("Invalid Git URL: URL too long or empty")
|
|
match = re.match(r"https?://[\w.\-]+(:\d+)?/(.+?)(?:\.git)?/?$", url)
|
|
if match:
|
|
full_path = match.group(2)
|
|
if "/" not in full_path:
|
|
raise ValueError(f"Cannot parse repository URL: {url}")
|
|
namespace, _, repo = full_path.rpartition("/")
|
|
return namespace, repo
|
|
match = re.match(r"git@[\w.\-]+:(.+?)(?:\.git)?$", url)
|
|
if match:
|
|
full_path = match.group(1)
|
|
if "/" not in full_path:
|
|
raise ValueError(f"Cannot parse repository URL: {url}")
|
|
namespace, _, repo = full_path.rpartition("/")
|
|
return namespace, repo
|
|
raise ValueError(f"Cannot parse repository URL: {url}")
|
|
|
|
async def test_connection(self, repo_url: str, token: str, client: httpx.AsyncClient) -> dict:
|
|
try:
|
|
owner, repo = self.parse_repo_url(repo_url)
|
|
api_base = self.get_api_base(repo_url)
|
|
headers = self.get_headers(token)
|
|
encoded_path = urllib.parse.quote(f"{owner}/{repo}", safe="")
|
|
|
|
response = await client.get(f"{api_base}/projects/{encoded_path}", headers=headers)
|
|
|
|
if response.status_code == 401:
|
|
return {"success": False, "message": "Invalid access token", "repo_name": None, "permissions": None}
|
|
if response.status_code == 404:
|
|
return {
|
|
"success": False,
|
|
"message": "Repository not found. Check URL and token permissions.",
|
|
"repo_name": None,
|
|
"permissions": None,
|
|
}
|
|
if response.status_code != 200:
|
|
return {
|
|
"success": False,
|
|
"message": f"API error: {response.status_code}",
|
|
"repo_name": None,
|
|
"permissions": None,
|
|
}
|
|
|
|
data = response.json()
|
|
perms = data.get("permissions") or {}
|
|
project_level = (perms.get("project_access") or {}).get("access_level", 0)
|
|
group_level = (perms.get("group_access") or {}).get("access_level", 0)
|
|
effective = max(project_level, group_level)
|
|
|
|
# GitLab uses visibility="private" / "internal" / "public". Both
|
|
# "internal" (signed-in users) and "public" are non-private for
|
|
# the purposes of this safety check.
|
|
visibility = (data.get("visibility") or "").lower()
|
|
is_private = visibility == "private"
|
|
|
|
if effective < 30: # Developer = 30, Maintainer = 40, Owner = 50
|
|
return {
|
|
"success": False,
|
|
"message": "Token requires Developer access or higher to push",
|
|
"repo_name": data.get("name_with_namespace"),
|
|
"permissions": perms,
|
|
"is_private": is_private,
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Connection successful",
|
|
"repo_name": data.get("name_with_namespace"),
|
|
"permissions": perms,
|
|
"is_private": is_private,
|
|
}
|
|
except Exception as e:
|
|
logger.error("GitLab connection test failed: %s", e)
|
|
return {
|
|
"success": False,
|
|
"message": f"Connection failed: {type(e).__name__}",
|
|
"repo_name": None,
|
|
"permissions": None,
|
|
"is_private": None,
|
|
}
|
|
|
|
async def push_files(
|
|
self,
|
|
repo_url: str,
|
|
token: str,
|
|
branch: str,
|
|
files: dict,
|
|
client: httpx.AsyncClient,
|
|
) -> dict:
|
|
try:
|
|
owner, repo = self.parse_repo_url(repo_url)
|
|
api_base = self.get_api_base(repo_url)
|
|
headers = self.get_headers(token)
|
|
encoded_path = urllib.parse.quote(f"{owner}/{repo}", safe="")
|
|
|
|
encoded_branch = urllib.parse.quote(branch, safe="")
|
|
branch_response = await client.get(
|
|
f"{api_base}/projects/{encoded_path}/repository/branches/{encoded_branch}",
|
|
headers=headers,
|
|
)
|
|
|
|
if branch_response.status_code == 404:
|
|
proj_response = await client.get(f"{api_base}/projects/{encoded_path}", headers=headers)
|
|
if proj_response.status_code != 200:
|
|
return {"status": "failed", "message": "Failed to get project info"}
|
|
|
|
default_branch = proj_response.json().get("default_branch", "main")
|
|
default_encoded = urllib.parse.quote(default_branch, safe="")
|
|
default_response = await client.get(
|
|
f"{api_base}/projects/{encoded_path}/repository/branches/{default_encoded}",
|
|
headers=headers,
|
|
)
|
|
|
|
if default_response.status_code != 200:
|
|
return await self._create_initial_commit(client, headers, api_base, encoded_path, branch, files)
|
|
|
|
create_response = await client.post(
|
|
f"{api_base}/projects/{encoded_path}/repository/branches",
|
|
headers=headers,
|
|
json={"branch": branch, "ref": default_branch},
|
|
)
|
|
if create_response.status_code not in (200, 201):
|
|
return {"status": "failed", "message": f"Failed to create branch: {create_response.status_code}"}
|
|
elif branch_response.status_code != 200:
|
|
return {"status": "failed", "message": f"Failed to check branch: {branch_response.status_code}"}
|
|
|
|
existing_blobs: dict[str, str] = {}
|
|
page = 1
|
|
while True:
|
|
tree_response = await client.get(
|
|
f"{api_base}/projects/{encoded_path}/repository/tree",
|
|
headers=headers,
|
|
params={"recursive": "true", "ref": branch, "per_page": 100, "page": page},
|
|
)
|
|
if tree_response.status_code != 200:
|
|
break
|
|
items = tree_response.json()
|
|
if not items:
|
|
break
|
|
for item in items:
|
|
if item.get("type") == "blob":
|
|
existing_blobs[item["path"]] = item["id"]
|
|
page += 1
|
|
|
|
actions = []
|
|
for path, content in files.items():
|
|
content_str = json.dumps(content, indent=2, default=str)
|
|
content_bytes = content_str.encode("utf-8")
|
|
content_sha = self._blob_sha(content_bytes)
|
|
|
|
if path in existing_blobs and existing_blobs[path] == content_sha:
|
|
continue
|
|
|
|
actions.append(
|
|
{
|
|
"action": "update" if path in existing_blobs else "create",
|
|
"file_path": path,
|
|
"content": base64.b64encode(content_bytes).decode(),
|
|
"encoding": "base64",
|
|
}
|
|
)
|
|
|
|
if not actions:
|
|
return {"status": "skipped", "message": "No changes to commit", "commit_sha": None, "files_changed": 0}
|
|
|
|
commit_message = f"Bambuddy backup - {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}"
|
|
commit_response = await client.post(
|
|
f"{api_base}/projects/{encoded_path}/repository/commits",
|
|
headers=headers,
|
|
json={"branch": branch, "commit_message": commit_message, "actions": actions},
|
|
)
|
|
if commit_response.status_code not in (200, 201):
|
|
return {
|
|
"status": "failed",
|
|
"message": f"Failed to create commit: {self._truncated_response_text(commit_response)}",
|
|
}
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": f"Backup successful - {len(actions)} files updated",
|
|
"commit_sha": commit_response.json().get("id"),
|
|
"files_changed": len(actions),
|
|
}
|
|
except Exception as e:
|
|
logger.error("Push to GitLab failed: %s", e)
|
|
return {"status": "failed", "message": str(e), "error": str(e)}
|
|
|
|
async def _create_initial_commit(
|
|
self,
|
|
client: httpx.AsyncClient,
|
|
headers: dict,
|
|
api_base: str,
|
|
encoded_path: str,
|
|
branch: str,
|
|
files: dict,
|
|
) -> dict:
|
|
"""Create the first commit in an empty repository."""
|
|
try:
|
|
actions = []
|
|
for path, content in files.items():
|
|
content_str = json.dumps(content, indent=2, default=str)
|
|
actions.append(
|
|
{
|
|
"action": "create",
|
|
"file_path": path,
|
|
"content": base64.b64encode(content_str.encode()).decode(),
|
|
"encoding": "base64",
|
|
}
|
|
)
|
|
|
|
commit_message = f"Initial Bambuddy backup - {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}"
|
|
commit_response = await client.post(
|
|
f"{api_base}/projects/{encoded_path}/repository/commits",
|
|
headers=headers,
|
|
json={"branch": branch, "commit_message": commit_message, "actions": actions, "start_branch": branch},
|
|
)
|
|
if commit_response.status_code not in (200, 201):
|
|
return {
|
|
"status": "failed",
|
|
"message": f"Failed to create initial commit: {self._truncated_response_text(commit_response)}",
|
|
}
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": f"Initial backup created - {len(files)} files",
|
|
"commit_sha": commit_response.json().get("id"),
|
|
"files_changed": len(files),
|
|
}
|
|
except Exception as e:
|
|
return {"status": "failed", "message": str(e)}
|