|
| 1 | +#!/usr/bin/env python3 |
| 2 | +""" |
| 3 | +License validation for CAT. |
| 4 | +Verifies Ed25519-signed JWT tokens and extracts tier information. |
| 5 | +""" |
| 6 | +from __future__ import annotations |
| 7 | + |
| 8 | +import json |
| 9 | +import base64 |
| 10 | +import sys |
| 11 | +from pathlib import Path |
| 12 | +from datetime import datetime, timezone |
| 13 | +from typing import TYPE_CHECKING, Any |
| 14 | + |
| 15 | +# Try to import cryptography, fail gracefully if not available |
| 16 | +try: |
| 17 | + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey |
| 18 | + from cryptography.hazmat.primitives.serialization import load_pem_public_key |
| 19 | + from cryptography.exceptions import InvalidSignature |
| 20 | + CRYPTO_AVAILABLE = True |
| 21 | +except ImportError: |
| 22 | + CRYPTO_AVAILABLE = False |
| 23 | + if TYPE_CHECKING: |
| 24 | + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey |
| 25 | + |
| 26 | + |
| 27 | +def base64url_decode(data: str) -> bytes: |
| 28 | + """Decode base64url without padding.""" |
| 29 | + padding = 4 - len(data) % 4 |
| 30 | + if padding != 4: |
| 31 | + data += '=' * padding |
| 32 | + return base64.urlsafe_b64decode(data) |
| 33 | + |
| 34 | + |
| 35 | +def load_public_key(key_path: Path) -> Any: |
| 36 | + """Load Ed25519 public key from PEM file.""" |
| 37 | + with open(key_path, 'rb') as f: |
| 38 | + return load_pem_public_key(f.read()) |
| 39 | + |
| 40 | + |
| 41 | +def validate_token(token: str, public_key: Any) -> dict: |
| 42 | + """ |
| 43 | + Validate JWT token and return payload. |
| 44 | +
|
| 45 | + Returns dict with: |
| 46 | + - valid: bool - whether signature is valid |
| 47 | + - tier: str - license tier (or "indie" if invalid) |
| 48 | + - expired: bool - whether token is past expiration |
| 49 | + - inGrace: bool - whether in grace period |
| 50 | + - daysRemaining: int - days until expiration (negative if expired) |
| 51 | + - error: str | None - error message if any |
| 52 | + """ |
| 53 | + result = { |
| 54 | + "valid": False, |
| 55 | + "tier": "indie", |
| 56 | + "expired": False, |
| 57 | + "inGrace": False, |
| 58 | + "daysRemaining": 0, |
| 59 | + "error": None |
| 60 | + } |
| 61 | + |
| 62 | + try: |
| 63 | + # Split JWT |
| 64 | + parts = token.split('.') |
| 65 | + if len(parts) != 3: |
| 66 | + result["error"] = "Invalid token format" |
| 67 | + return result |
| 68 | + |
| 69 | + header_b64, payload_b64, signature_b64 = parts |
| 70 | + |
| 71 | + # Verify signature |
| 72 | + message = f"{header_b64}.{payload_b64}".encode('utf-8') |
| 73 | + signature = base64url_decode(signature_b64) |
| 74 | + |
| 75 | + try: |
| 76 | + public_key.verify(signature, message) |
| 77 | + except InvalidSignature: |
| 78 | + result["error"] = "Invalid signature" |
| 79 | + return result |
| 80 | + |
| 81 | + # Decode payload |
| 82 | + payload_json = base64url_decode(payload_b64).decode('utf-8') |
| 83 | + payload = json.loads(payload_json) |
| 84 | + |
| 85 | + # Extract fields |
| 86 | + tier = payload.get("tier", "indie") |
| 87 | + exp = payload.get("exp") # Unix timestamp |
| 88 | + grace_days = payload.get("grace_days", payload.get("graceDays", 7)) |
| 89 | + |
| 90 | + result["valid"] = True |
| 91 | + result["tier"] = tier |
| 92 | + |
| 93 | + # Check expiration |
| 94 | + if exp: |
| 95 | + exp_dt = datetime.fromtimestamp(exp, tz=timezone.utc) |
| 96 | + now = datetime.now(timezone.utc) |
| 97 | + delta = exp_dt - now |
| 98 | + result["daysRemaining"] = delta.days |
| 99 | + |
| 100 | + if now > exp_dt: |
| 101 | + result["expired"] = True |
| 102 | + # Check grace period |
| 103 | + grace_delta = (now - exp_dt).days |
| 104 | + if grace_delta <= grace_days: |
| 105 | + result["inGrace"] = True |
| 106 | + else: |
| 107 | + # Past grace period - fall back to indie |
| 108 | + result["tier"] = "indie" |
| 109 | + |
| 110 | + return result |
| 111 | + |
| 112 | + except Exception as e: |
| 113 | + result["error"] = str(e) |
| 114 | + return result |
| 115 | + |
| 116 | + |
| 117 | +def find_config_file() -> Path | None: |
| 118 | + """Find cat-config.local.json in project directory.""" |
| 119 | + # Check current directory and parents for .claude/cat/ |
| 120 | + current = Path.cwd() |
| 121 | + for parent in [current] + list(current.parents): |
| 122 | + config_path = parent / ".claude" / "cat" / "cat-config.local.json" |
| 123 | + if config_path.exists(): |
| 124 | + return config_path |
| 125 | + return None |
| 126 | + |
| 127 | + |
| 128 | +def find_public_key() -> Path | None: |
| 129 | + """Find public key file.""" |
| 130 | + # Check plugin config directory |
| 131 | + script_dir = Path(__file__).parent |
| 132 | + key_path = script_dir.parent / "config" / "cat-public-key.pem" |
| 133 | + if key_path.exists(): |
| 134 | + return key_path |
| 135 | + |
| 136 | + # Check CLAUDE_PLUGIN_ROOT |
| 137 | + import os |
| 138 | + plugin_root = os.environ.get("CLAUDE_PLUGIN_ROOT") |
| 139 | + if plugin_root: |
| 140 | + key_path = Path(plugin_root) / "config" / "cat-public-key.pem" |
| 141 | + if key_path.exists(): |
| 142 | + return key_path |
| 143 | + |
| 144 | + return None |
| 145 | + |
| 146 | + |
| 147 | +def main(): |
| 148 | + """Main entry point.""" |
| 149 | + result = { |
| 150 | + "valid": False, |
| 151 | + "tier": "indie", |
| 152 | + "expired": False, |
| 153 | + "inGrace": False, |
| 154 | + "daysRemaining": 0, |
| 155 | + "error": None, |
| 156 | + "warning": None |
| 157 | + } |
| 158 | + |
| 159 | + # Check if cryptography is available |
| 160 | + if not CRYPTO_AVAILABLE: |
| 161 | + result["error"] = "cryptography library not available" |
| 162 | + result["warning"] = "Install with: pip install cryptography" |
| 163 | + print(json.dumps(result)) |
| 164 | + return 1 |
| 165 | + |
| 166 | + # Find config file |
| 167 | + config_path = find_config_file() |
| 168 | + if not config_path: |
| 169 | + # No config = free tier, not an error |
| 170 | + print(json.dumps(result)) |
| 171 | + return 0 |
| 172 | + |
| 173 | + # Read config |
| 174 | + try: |
| 175 | + with open(config_path) as f: |
| 176 | + config = json.load(f) |
| 177 | + except Exception as e: |
| 178 | + result["error"] = f"Failed to read config: {e}" |
| 179 | + print(json.dumps(result)) |
| 180 | + return 1 |
| 181 | + |
| 182 | + # Get token |
| 183 | + token = config.get("license") |
| 184 | + if not token: |
| 185 | + # No license = free tier |
| 186 | + print(json.dumps(result)) |
| 187 | + return 0 |
| 188 | + |
| 189 | + # Find public key |
| 190 | + key_path = find_public_key() |
| 191 | + if not key_path: |
| 192 | + result["error"] = "Public key not found" |
| 193 | + print(json.dumps(result)) |
| 194 | + return 1 |
| 195 | + |
| 196 | + # Load public key and validate |
| 197 | + try: |
| 198 | + public_key = load_public_key(key_path) |
| 199 | + result = validate_token(token, public_key) |
| 200 | + except Exception as e: |
| 201 | + result["error"] = f"Validation failed: {e}" |
| 202 | + |
| 203 | + # Add warning for grace period |
| 204 | + if result["inGrace"]: |
| 205 | + days_past = abs(result["daysRemaining"]) |
| 206 | + result["warning"] = f"License expired {days_past} days ago. Renew soon." |
| 207 | + elif result["expired"] and not result["inGrace"]: |
| 208 | + result["warning"] = "License expired. Run /cat:login to renew." |
| 209 | + |
| 210 | + print(json.dumps(result)) |
| 211 | + return 0 if result["valid"] else 1 |
| 212 | + |
| 213 | + |
| 214 | +if __name__ == "__main__": |
| 215 | + sys.exit(main()) |
0 commit comments