Spaces:
Sleeping
Sleeping
| """ | |
| GitHub Storage Module | |
| Handles encrypted read/write operations to a GitHub repository. | |
| """ | |
| import os | |
| import json | |
| import base64 | |
| import requests | |
| from cryptography.fernet import Fernet | |
| # Configuration from environment variables | |
| GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "") | |
| ENCRYPTION_KEY = os.environ.get("ENCRYPTION_KEY", "") | |
| GITHUB_REPO = os.environ.get("GITHUB_REPO", "") | |
| DATA_PATH = "data/users.enc" | |
| def get_cipher(): | |
| """Get Fernet cipher instance.""" | |
| if not ENCRYPTION_KEY: | |
| raise ValueError("ENCRYPTION_KEY environment variable not set") | |
| return Fernet(ENCRYPTION_KEY.encode()) | |
| def read_users() -> dict: | |
| """Read and decrypt user data from GitHub.""" | |
| if not GITHUB_TOKEN or not GITHUB_REPO: | |
| raise ValueError("GITHUB_TOKEN and GITHUB_REPO must be set") | |
| url = f"https://api.github.com/repos/{GITHUB_REPO}/contents/{DATA_PATH}" | |
| headers = { | |
| "Authorization": f"token {GITHUB_TOKEN}", | |
| "Accept": "application/vnd.github.v3+json" | |
| } | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 404: | |
| # No data file yet, return empty users dict | |
| return {"users": {}} | |
| if response.status_code != 200: | |
| raise Exception(f"GitHub API error: {response.status_code} - {response.text}") | |
| content = response.json() | |
| encrypted_data = base64.b64decode(content["content"]) | |
| # Handle initial placeholder | |
| try: | |
| cipher = get_cipher() | |
| decrypted = cipher.decrypt(encrypted_data) | |
| return json.loads(decrypted) | |
| except Exception: | |
| # If decryption fails (e.g., placeholder data), return empty | |
| return {"users": {}} | |
| def write_users(data: dict) -> bool: | |
| """Encrypt and write user data to GitHub.""" | |
| if not GITHUB_TOKEN or not GITHUB_REPO: | |
| raise ValueError("GITHUB_TOKEN and GITHUB_REPO must be set") | |
| url = f"https://api.github.com/repos/{GITHUB_REPO}/contents/{DATA_PATH}" | |
| headers = { | |
| "Authorization": f"token {GITHUB_TOKEN}", | |
| "Accept": "application/vnd.github.v3+json" | |
| } | |
| # Encrypt the data | |
| cipher = get_cipher() | |
| encrypted = cipher.encrypt(json.dumps(data).encode()) | |
| content_b64 = base64.b64encode(encrypted).decode() | |
| # Get current file SHA (required for updates) | |
| get_response = requests.get(url, headers=headers) | |
| sha = None | |
| if get_response.status_code == 200: | |
| sha = get_response.json().get("sha") | |
| # Create or update file | |
| payload = { | |
| "message": "Update user data", | |
| "content": content_b64, | |
| "branch": "main" | |
| } | |
| if sha: | |
| payload["sha"] = sha | |
| response = requests.put(url, headers=headers, json=payload) | |
| return response.status_code in [200, 201] | |
| def get_file_sha(path: str) -> str | None: | |
| """Get the SHA of a file in the repo.""" | |
| url = f"https://api.github.com/repos/{GITHUB_REPO}/contents/{path}" | |
| headers = { | |
| "Authorization": f"token {GITHUB_TOKEN}", | |
| "Accept": "application/vnd.github.v3+json" | |
| } | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| return response.json().get("sha") | |
| return None | |