| |
|
|
| import os |
| import json |
| from pathlib import Path |
| from cryptography.fernet import Fernet |
|
|
| STORAGE_DIR = Path("oauth_tokens") |
| STORAGE_DIR.mkdir(exist_ok=True) |
|
|
| KEY_PATH = STORAGE_DIR / "secret.key" |
|
|
|
|
| def load_key(): |
| if not KEY_PATH.exists(): |
| key = Fernet.generate_key() |
| KEY_PATH.write_bytes(key) |
| return KEY_PATH.read_bytes() |
|
|
|
|
| fernet = Fernet(load_key()) |
|
|
|
|
| def _file(user_id, provider): |
| return STORAGE_DIR / f"{user_id}_{provider}.json" |
|
|
|
|
| def save_tokens(user_id: str, provider: str, data: dict): |
| encrypted = fernet.encrypt(json.dumps(data).encode()) |
| _file(user_id, provider).write_bytes(encrypted) |
|
|
|
|
| def load_tokens(user_id: str, provider: str): |
| f = _file(user_id, provider) |
| if not f.exists(): |
| return None |
| decrypted = fernet.decrypt(f.read_bytes()) |
| return json.loads(decrypted.decode()) |
|
|
|
|
| def delete_tokens(user_id: str, provider: str): |
| f = _file(user_id, provider) |
| if f.exists(): |
| f.unlink() |
