forked from claw/flaskpaste
fpaste: add E2E encryption support
-e/--encrypt flag encrypts content with AES-256-GCM before upload. Key is appended to URL fragment (#...), never sent to server. Auto-detects key fragment on retrieval and decrypts locally.
This commit is contained in:
93
fpaste
93
fpaste
@@ -2,6 +2,7 @@
|
|||||||
"""FlaskPaste command-line client."""
|
"""FlaskPaste command-line client."""
|
||||||
|
|
||||||
import argparse
|
import argparse
|
||||||
|
import base64
|
||||||
import hashlib
|
import hashlib
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
@@ -10,6 +11,14 @@ import urllib.error
|
|||||||
import urllib.request
|
import urllib.request
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
# Optional encryption support
|
||||||
|
try:
|
||||||
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||||
|
|
||||||
|
HAS_CRYPTO = True
|
||||||
|
except ImportError:
|
||||||
|
HAS_CRYPTO = False
|
||||||
|
|
||||||
|
|
||||||
def get_config():
|
def get_config():
|
||||||
"""Load configuration from environment or config file."""
|
"""Load configuration from environment or config file."""
|
||||||
@@ -55,6 +64,48 @@ def die(msg, code=1):
|
|||||||
sys.exit(code)
|
sys.exit(code)
|
||||||
|
|
||||||
|
|
||||||
|
def encrypt_content(plaintext):
|
||||||
|
"""Encrypt content with AES-256-GCM. Returns (ciphertext, key)."""
|
||||||
|
if not HAS_CRYPTO:
|
||||||
|
die("encryption requires 'cryptography' package: pip install cryptography")
|
||||||
|
key = os.urandom(32)
|
||||||
|
nonce = os.urandom(12) # 96-bit nonce for GCM
|
||||||
|
aesgcm = AESGCM(key)
|
||||||
|
ciphertext = aesgcm.encrypt(nonce, plaintext, None)
|
||||||
|
return nonce + ciphertext, key
|
||||||
|
|
||||||
|
|
||||||
|
def decrypt_content(blob, key):
|
||||||
|
"""Decrypt AES-256-GCM encrypted content."""
|
||||||
|
if not HAS_CRYPTO:
|
||||||
|
die("decryption requires 'cryptography' package: pip install cryptography")
|
||||||
|
if len(blob) < 12:
|
||||||
|
die("encrypted content too short")
|
||||||
|
nonce, ciphertext = blob[:12], blob[12:]
|
||||||
|
aesgcm = AESGCM(key)
|
||||||
|
try:
|
||||||
|
return aesgcm.decrypt(nonce, ciphertext, None)
|
||||||
|
except Exception:
|
||||||
|
die("decryption failed (wrong key or corrupted data)")
|
||||||
|
|
||||||
|
|
||||||
|
def encode_key(key):
|
||||||
|
"""Encode key as URL-safe base64."""
|
||||||
|
return base64.urlsafe_b64encode(key).decode().rstrip("=")
|
||||||
|
|
||||||
|
|
||||||
|
def decode_key(encoded):
|
||||||
|
"""Decode URL-safe base64 key."""
|
||||||
|
# Add padding if needed
|
||||||
|
padding = 4 - (len(encoded) % 4)
|
||||||
|
if padding != 4:
|
||||||
|
encoded += "=" * padding
|
||||||
|
try:
|
||||||
|
return base64.urlsafe_b64decode(encoded)
|
||||||
|
except Exception:
|
||||||
|
die("invalid encryption key in URL")
|
||||||
|
|
||||||
|
|
||||||
def solve_pow(nonce, difficulty):
|
def solve_pow(nonce, difficulty):
|
||||||
"""Solve proof-of-work challenge.
|
"""Solve proof-of-work challenge.
|
||||||
|
|
||||||
@@ -122,6 +173,15 @@ def cmd_create(args, config):
|
|||||||
if not content:
|
if not content:
|
||||||
die("empty content")
|
die("empty content")
|
||||||
|
|
||||||
|
# Encrypt content if requested
|
||||||
|
encryption_key = None
|
||||||
|
if args.encrypt:
|
||||||
|
if not args.quiet:
|
||||||
|
print("encrypting...", end="", file=sys.stderr)
|
||||||
|
content, encryption_key = encrypt_content(content)
|
||||||
|
if not args.quiet:
|
||||||
|
print(" done", file=sys.stderr)
|
||||||
|
|
||||||
headers = {}
|
headers = {}
|
||||||
if config["cert_sha1"]:
|
if config["cert_sha1"]:
|
||||||
headers["X-SSL-Client-SHA1"] = config["cert_sha1"]
|
headers["X-SSL-Client-SHA1"] = config["cert_sha1"]
|
||||||
@@ -133,7 +193,7 @@ def cmd_create(args, config):
|
|||||||
print(f"solving pow (difficulty={challenge['difficulty']})...", end="", file=sys.stderr)
|
print(f"solving pow (difficulty={challenge['difficulty']})...", end="", file=sys.stderr)
|
||||||
solution = solve_pow(challenge["nonce"], challenge["difficulty"])
|
solution = solve_pow(challenge["nonce"], challenge["difficulty"])
|
||||||
if not args.quiet:
|
if not args.quiet:
|
||||||
print(f" done", file=sys.stderr)
|
print(" done", file=sys.stderr)
|
||||||
headers["X-PoW-Token"] = challenge["token"]
|
headers["X-PoW-Token"] = challenge["token"]
|
||||||
headers["X-PoW-Solution"] = str(solution)
|
headers["X-PoW-Solution"] = str(solution)
|
||||||
|
|
||||||
@@ -142,12 +202,17 @@ def cmd_create(args, config):
|
|||||||
|
|
||||||
if status == 201:
|
if status == 201:
|
||||||
data = json.loads(body)
|
data = json.loads(body)
|
||||||
|
# Append encryption key to URL fragment if encrypted
|
||||||
|
key_fragment = ""
|
||||||
|
if encryption_key:
|
||||||
|
key_fragment = "#" + encode_key(encryption_key)
|
||||||
|
|
||||||
if args.raw:
|
if args.raw:
|
||||||
print(config["server"].rstrip("/") + data["raw"])
|
print(config["server"].rstrip("/") + data["raw"] + key_fragment)
|
||||||
elif args.quiet:
|
elif args.quiet:
|
||||||
print(data["id"])
|
print(data["id"] + key_fragment)
|
||||||
else:
|
else:
|
||||||
print(config["server"].rstrip("/") + data["url"])
|
print(config["server"].rstrip("/") + data["url"] + key_fragment)
|
||||||
else:
|
else:
|
||||||
try:
|
try:
|
||||||
err = json.loads(body).get("error", body.decode())
|
err = json.loads(body).get("error", body.decode())
|
||||||
@@ -158,7 +223,17 @@ def cmd_create(args, config):
|
|||||||
|
|
||||||
def cmd_get(args, config):
|
def cmd_get(args, config):
|
||||||
"""Retrieve a paste."""
|
"""Retrieve a paste."""
|
||||||
paste_id = args.id.split("/")[-1] # Handle full URLs
|
# Parse URL for paste ID and optional encryption key fragment
|
||||||
|
url_input = args.id
|
||||||
|
encryption_key = None
|
||||||
|
|
||||||
|
# Extract key from URL fragment (#...)
|
||||||
|
if "#" in url_input:
|
||||||
|
url_input, key_encoded = url_input.rsplit("#", 1)
|
||||||
|
if key_encoded:
|
||||||
|
encryption_key = decode_key(key_encoded)
|
||||||
|
|
||||||
|
paste_id = url_input.split("/")[-1] # Handle full URLs
|
||||||
base = config["server"].rstrip("/")
|
base = config["server"].rstrip("/")
|
||||||
|
|
||||||
if args.meta:
|
if args.meta:
|
||||||
@@ -170,12 +245,18 @@ def cmd_get(args, config):
|
|||||||
print(f"mime_type: {data['mime_type']}")
|
print(f"mime_type: {data['mime_type']}")
|
||||||
print(f"size: {data['size']}")
|
print(f"size: {data['size']}")
|
||||||
print(f"created_at: {data['created_at']}")
|
print(f"created_at: {data['created_at']}")
|
||||||
|
if encryption_key:
|
||||||
|
print(f"encrypted: yes (key in URL)")
|
||||||
else:
|
else:
|
||||||
die(f"not found: {paste_id}")
|
die(f"not found: {paste_id}")
|
||||||
else:
|
else:
|
||||||
url = f"{base}/{paste_id}/raw"
|
url = f"{base}/{paste_id}/raw"
|
||||||
status, body, headers = request(url)
|
status, body, headers = request(url)
|
||||||
if status == 200:
|
if status == 200:
|
||||||
|
# Decrypt if encryption key was provided
|
||||||
|
if encryption_key:
|
||||||
|
body = decrypt_content(body, encryption_key)
|
||||||
|
|
||||||
if args.output:
|
if args.output:
|
||||||
Path(args.output).write_bytes(body)
|
Path(args.output).write_bytes(body)
|
||||||
print(f"saved: {args.output}", file=sys.stderr)
|
print(f"saved: {args.output}", file=sys.stderr)
|
||||||
@@ -241,6 +322,7 @@ def main():
|
|||||||
# create
|
# create
|
||||||
p_create = subparsers.add_parser("create", aliases=["c", "new"], help="create paste")
|
p_create = subparsers.add_parser("create", aliases=["c", "new"], help="create paste")
|
||||||
p_create.add_argument("file", nargs="?", help="file to upload (- for stdin)")
|
p_create.add_argument("file", nargs="?", help="file to upload (- for stdin)")
|
||||||
|
p_create.add_argument("-e", "--encrypt", action="store_true", help="encrypt content (E2E)")
|
||||||
p_create.add_argument("-r", "--raw", action="store_true", help="output raw URL")
|
p_create.add_argument("-r", "--raw", action="store_true", help="output raw URL")
|
||||||
p_create.add_argument("-q", "--quiet", action="store_true", help="output ID only")
|
p_create.add_argument("-q", "--quiet", action="store_true", help="output ID only")
|
||||||
|
|
||||||
@@ -268,6 +350,7 @@ def main():
|
|||||||
if not sys.stdin.isatty():
|
if not sys.stdin.isatty():
|
||||||
args.command = "create"
|
args.command = "create"
|
||||||
args.file = None
|
args.file = None
|
||||||
|
args.encrypt = False
|
||||||
args.raw = False
|
args.raw = False
|
||||||
args.quiet = False
|
args.quiet = False
|
||||||
else:
|
else:
|
||||||
|
|||||||
Reference in New Issue
Block a user