""" Share Token Management for NoteDiscovery Handles creating, storing, and revoking share tokens for public note access. """ import json import logging import secrets import string from pathlib import Path from datetime import datetime, timezone from typing import Optional, Dict, Any import threading from .utils import validate_path_security logger = logging.getLogger("uvicorn.error") # Thread lock for safe concurrent access _lock = threading.Lock() def _token_references_accessible_file(storage_dir: str, rel_path: Any) -> bool: """ Return True if rel_path is a string that resolves to a regular file under storage_dir and passes the same path boundary check used elsewhere in the app. """ if not rel_path or not isinstance(rel_path, str): return False try: full = (Path(storage_dir) / rel_path).resolve() except (OSError, ValueError, RuntimeError): return False if not validate_path_security(str(storage_dir), full): return False return full.is_file() def _prune_inaccessible_unsafe(data_dir: str) -> int: """ Remove share tokens whose stored path is missing or not under the notes directory. Call only while holding _lock. Returns the number of tokens removed. """ tokens = load_tokens(data_dir) if not tokens: return 0 kept: Dict[str, Dict[str, Any]] = {} for token, info in tokens.items(): path = info.get("path") if _token_references_accessible_file(data_dir, path): kept[token] = info removed = len(tokens) - len(kept) if removed and not save_tokens(data_dir, kept): return 0 return removed def prune_inaccessible_share_tokens(data_dir: str) -> int: """ Remove entries whose path no longer resolves to a file under the storage root. Called automatically after create/revoke share; may also be used from tests or one-off maintenance. """ with _lock: return _prune_inaccessible_unsafe(data_dir) def generate_token(length: int = 16) -> str: """Generate a URL-safe random token.""" # Use alphanumeric + underscore/hyphen (URL-safe) alphabet = string.ascii_letters + string.digits + '_-' return ''.join(secrets.choice(alphabet) for _ in range(length)) def get_tokens_file_path(data_dir: str) -> Path: """Get the path to the share tokens file.""" return Path(data_dir) / '.share-tokens.json' def load_tokens(data_dir: str) -> Dict[str, Dict[str, Any]]: """Load share tokens from file.""" tokens_file = get_tokens_file_path(data_dir) if not tokens_file.exists(): return {} try: with open(tokens_file, 'r', encoding='utf-8') as f: return json.load(f) except (json.JSONDecodeError, IOError): return {} def save_tokens(data_dir: str, tokens: Dict[str, Dict[str, Any]]) -> bool: """Save share tokens to file.""" tokens_file = get_tokens_file_path(data_dir) try: # Ensure parent directory exists tokens_file.parent.mkdir(parents=True, exist_ok=True) with open(tokens_file, 'w', encoding='utf-8') as f: json.dump(tokens, f, indent=2, ensure_ascii=False) return True except IOError as e: logger.error("Failed to save share tokens: %s", e) return False def create_share_token(data_dir: str, note_path: str, theme: str = "light") -> Optional[str]: """ Create a share token for a note. If the note already has a token, returns the existing one. Args: data_dir: Path to the data directory note_path: Path to the note (relative to notes_dir) theme: The theme to use when viewing the shared note Returns: The share token, or None on error """ with _lock: tokens = load_tokens(data_dir) # Check if note already has a token for token, info in tokens.items(): if info.get('path') == note_path: _prune_inaccessible_unsafe(data_dir) return token # Generate new token token = generate_token() # Ensure uniqueness (extremely unlikely collision, but check anyway) while token in tokens: token = generate_token() # Store token with theme tokens[token] = { 'path': note_path, 'theme': theme, 'created': datetime.now(timezone.utc).isoformat() } if save_tokens(data_dir, tokens): _prune_inaccessible_unsafe(data_dir) return token _prune_inaccessible_unsafe(data_dir) return None def get_share_token(data_dir: str, note_path: str) -> Optional[str]: """ Get the share token for a note, if it exists. Args: data_dir: Path to the data directory note_path: Path to the note Returns: The share token, or None if not shared """ tokens = load_tokens(data_dir) for token, info in tokens.items(): if info.get('path') == note_path: return token return None def get_note_by_token(data_dir: str, token: str) -> Optional[Dict[str, Any]]: """ Get the note info for a share token. Args: data_dir: Path to the data directory token: The share token Returns: Dict with 'path' and 'theme', or None if token not found """ tokens = load_tokens(data_dir) if token in tokens: return { 'path': tokens[token].get('path'), 'theme': tokens[token].get('theme', 'light') } return None def get_all_shared_paths(data_dir: str) -> list: """ Get a list of all currently shared note paths. Used for displaying share indicators in the UI. Args: data_dir: Path to the data directory Returns: List of note paths that are currently shared """ tokens = load_tokens(data_dir) return [info.get('path') for info in tokens.values() if info.get('path')] def revoke_share_token(data_dir: str, note_path: str) -> bool: """ Revoke (delete) the share token for a note. Args: data_dir: Path to the data directory note_path: Path to the note Returns: True if token was revoked, False if not found or error """ with _lock: tokens = load_tokens(data_dir) # Find and remove token for this note token_to_remove = None for token, info in tokens.items(): if info.get('path') == note_path: token_to_remove = token break if token_to_remove: del tokens[token_to_remove] if not save_tokens(data_dir, tokens): _prune_inaccessible_unsafe(data_dir) return False _prune_inaccessible_unsafe(data_dir) return True _prune_inaccessible_unsafe(data_dir) return False def get_share_info(data_dir: str, note_path: str) -> Optional[Dict[str, Any]]: """ Get share information for a note. Args: data_dir: Path to the data directory note_path: Path to the note Returns: Dict with token, theme, and created date, or None if not shared """ tokens = load_tokens(data_dir) for token, info in tokens.items(): if info.get('path') == note_path: return { 'token': token, 'theme': info.get('theme', 'light'), 'created': info.get('created'), 'shared': True } return {'shared': False} def update_token_path(data_dir: str, old_path: str, new_path: str) -> bool: """ Update the path for a token when a note is moved/renamed. Args: data_dir: Path to the data directory old_path: Old note path new_path: New note path Returns: True if updated, False if not found or error """ with _lock: tokens = load_tokens(data_dir) for token, info in tokens.items(): if info.get('path') == old_path: info['path'] = new_path return save_tokens(data_dir, tokens) return False def delete_token_for_note(data_dir: str, note_path: str) -> bool: """ Delete the share token when a note is deleted. Alias for revoke_share_token for clarity. """ return revoke_share_token(data_dir, note_path)