paste-bin/app.py

import os
import uuid
import json
import fcntl
import secrets
import time
import threading
from datetime import datetime, timedelta, timezone
from flask import Flask, request, redirect, url_for, render_template, send_file, abort, make_response
from werkzeug.utils import secure_filename

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 20 * 1024 * 1024
app.config['UPLOAD_FOLDER'] = os.environ.get('UPLOAD_FOLDER', os.path.join(os.path.dirname(os.path.abspath(__file__)), 'uploads'))
app.config['STORE_FOLDER'] = os.environ.get('STORE_FOLDER', os.path.join(os.path.dirname(os.path.abspath(__file__)), 'store'))
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY') or secrets.token_hex(32)

EXPIRY_OPTIONS = [
    ('1h', '1 hour'),
    ('1d', '1 day'),
    ('1w', '1 week'),
    ('1m', '1 month'),
    ('forever', 'Never'),
]

ALLOWED_IMAGE_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp', 'tiff'}
ALLOWED_IMAGE_MAGIC = {
    'png': b'\x89PNG\r\n\x1a\n',
    'jpg': b'\xff\xd8\xff',
    'gif': b'GIF87a', b'GIF89a',
    'webp': b'RIFF',
    'bmp': b'BM',
}
ALLOWED_TEXT_EXTENSIONS = {'txt', 'py', 'js', 'ts', 'c', 'cpp', 'h', 'java', 'rb', 'go', 'rs', 'md', 'json', 'xml', 'yaml', 'yml', 'html', 'css', 'sh', 'log', 'csv', 'sql', 'ini', 'cfg', 'toml', 'lua', 'php', 'swift', 'kt', 'scala', 'r', 'pl', 'hs', 'zig', 'nix'}

_upload_attempts = {}
_UPLOAD_MAX = 10
_UPLOAD_WINDOW = 60

_csrf_secret = secrets.token_hex(32)

EXPIRY_OPTIONS = [
    ('1h', '1 hour'),
    ('1d', '1 day'),
    ('1w', '1 week'),
    ('1m', '1 month'),
    ('forever', 'Never'),
]

ALLOWED_IMAGE_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp', 'svg', 'tiff'}
ALLOWED_TEXT_EXTENSIONS = {'txt', 'py', 'js', 'ts', 'c', 'cpp', 'h', 'java', 'rb', 'go', 'rs', 'md', 'json', 'xml', 'yaml', 'yml', 'html', 'css', 'sh', 'log', 'csv', 'sql', 'ini', 'cfg', 'toml', 'lua', 'php', 'swift', 'kt', 'scala', 'r', 'pl', 'hs', 'zig', 'nix'}


def ensure_dirs():
    os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
    os.makedirs(app.config['STORE_FOLDER'], exist_ok=True)


def generate_id():
    return uuid.uuid4().hex[:16]


def parse_expiry(expiry_key):
    if expiry_key == 'forever':
        return None
    now = datetime.now(timezone.utc)
    if expiry_key == '1h':
        return (now + timedelta(hours=1)).isoformat()
    elif expiry_key == '1d':
        return (now + timedelta(days=1)).isoformat()
    elif expiry_key == '1w':
        return (now + timedelta(weeks=1)).isoformat()
    elif expiry_key == '1m':
        return (now + timedelta(days=30)).isoformat()
    return None


def store_paste(paste_id, paste_data):
    store_path = os.path.join(app.config['STORE_FOLDER'], paste_id)
    tmp_path = store_path + f".tmp.{os.getpid()}"
    fd = os.open(tmp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
    with os.fdopen(fd, 'w') as f:
        json.dump(paste_data, f)
    os.rename(tmp_path, store_path)


def load_paste(paste_id):
    store_path = os.path.join(app.config['STORE_FOLDER'], paste_id)
    if not os.path.exists(store_path):
        return None
    try:
        with open(store_path, 'r') as f:
            return json.load(f)
    except (json.JSONDecodeError, IOError):
        return None


def delete_paste(paste_id):
    store_path = os.path.join(app.config['STORE_FOLDER'], paste_id)
    txt_path = store_path + '.txt'
    paste = load_paste(paste_id)
    if paste and paste['type'] in ('image', 'file') and paste.get('filepath'):
        filepath = paste['filepath']
        if os.path.exists(filepath):
            os.remove(filepath)
    if os.path.exists(store_path):
        os.remove(store_path)
    if os.path.exists(txt_path):
        os.remove(txt_path)


def is_expired(paste):
    if not paste.get('expires_at'):
        return False
    expires_at = datetime.fromisoformat(paste['expires_at'])
    return expires_at < datetime.now(timezone.utc)


def cleanup_expired():
    for filename in os.listdir(app.config['STORE_FOLDER']):
        if filename.endswith('.txt'):
            continue
        filepath = os.path.join(app.config['STORE_FOLDER'], filename)
        try:
            with open(filepath, 'r') as f:
                paste = json.load(f)
        except (json.JSONDecodeError, IOError):
            continue
        if is_expired(paste):
            delete_paste(filename)


def _cleanup_loop():
    while True:
        time.sleep(300)
        try:
            cleanup_expired()
        except Exception:
            pass


threading.Thread(target=_cleanup_loop, daemon=True).start()


def is_image(filename):
    ext = filename.rsplit('.', 1)[-1].lower() if '.' in filename else ''
    return ext in ALLOWED_IMAGE_EXTENSIONS


def is_text_file(filename):
    ext = filename.rsplit('.', 1)[-1].lower() if '.' in filename else ''
    return ext in ALLOWED_TEXT_EXTENSIONS


def get_text_content(paste_id):
    content_path = os.path.join(app.config['STORE_FOLDER'], paste_id + '.txt')
    if os.path.exists(content_path):
        with open(content_path, 'r', encoding='utf-8') as f:
            return f.read()
    return None


def save_text_content(paste_id, content):
    content_path = os.path.join(app.config['STORE_FOLDER'], paste_id + '.txt')
    with open(content_path, 'w', encoding='utf-8') as f:
        f.write(content)


@app.before_request
def before_request():
    ensure_dirs()


@app.after_request
def add_security_headers(response):
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
    response.headers['Content-Security-Policy'] = "default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'"
    if request.secure:
        response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
    return response


def _check_upload_rate():
    ip = request.remote_addr or 'unknown'
    now = time.time()
    if ip not in _upload_attempts:
        _upload_attempts[ip] = []
    _upload_attempts[ip] = [t for t in _upload_attempts[ip] if now - t < _UPLOAD_WINDOW]
    if len(_upload_attempts[ip]) >= _UPLOAD_MAX:
        return False
    _upload_attempts[ip].append(now)
    return True


def _csrf_token():
    sess = request.cookies.get('csrf_token')
    if not sess:
        return secrets.token_hex(16)
    return sess


@app.route('/', methods=['GET'])
def index():
    resp = make_response(render_template('index.html', expiry_options=EXPIRY_OPTIONS, csrf_token=_csrf_token()))
    resp.set_cookie('csrf_token', _csrf_token(), httponly=False, samesite='Strict', path='/')
    return resp


@app.route('/paste', methods=['POST'])
def create_paste():
    if not _check_upload_rate():
        return render_template('index.html', expiry_options=EXPIRY_OPTIONS,
                               error='Too many uploads. Please wait.'), 429
    paste_type = request.form.get('paste_type', 'text')
    title = request.form.get('title', '').strip()
    expiry_key = request.form.get('expiry', '1d')
    expires_at = parse_expiry(expiry_key)
    created_at = datetime.now(timezone.utc).isoformat()

    if paste_type == 'text':
        content = request.form.get('content', '').strip()
        if not content:
            return render_template('index.html', expiry_options=EXPIRY_OPTIONS, error='Content cannot be empty'), 400

        paste_id = generate_id()
        save_text_content(paste_id, content)
        store_paste(paste_id, {
            'type': 'text',
            'title': title,
            'expires_at': expires_at,
            'created_at': created_at,
        })
        return redirect(url_for('view_paste', paste_id=paste_id))

    elif paste_type == 'image':
        file = request.files.get('file')
        if not file or file.filename == '':
            return render_template('index.html', expiry_options=EXPIRY_OPTIONS, error='No file selected'), 400

        filename = secure_filename(file.filename)
        if not is_image(filename):
            return render_template('index.html', expiry_options=EXPIRY_OPTIONS, error='Invalid image format. SVG not allowed. Allowed: ' + ', '.join(sorted(ALLOWED_IMAGE_EXTENSIONS))), 400

        head = file.read(12)
        file.seek(0)
        ext = filename.rsplit('.', 1)[-1].lower()
        if ext in ALLOWED_IMAGE_MAGIC:
            valid = any(head.startswith(m) for m in (ALLOWED_IMAGE_MAGIC[ext] if isinstance(ALLOWED_IMAGE_MAGIC[ext], tuple) else (ALLOWED_IMAGE_MAGIC[ext],)))
            if not valid:
                return render_template('index.html', expiry_options=EXPIRY_OPTIONS,
                                       error='File content does not match image type.'), 400

        paste_id = generate_id()
        ext = filename.rsplit('.', 1)[-1]
        saved_name = f"{paste_id}.{ext}"
        filepath = os.path.join(app.config['UPLOAD_FOLDER'], saved_name)
        file.save(filepath)

        store_paste(paste_id, {
            'type': 'image',
            'title': title or filename,
            'expires_at': expires_at,
            'created_at': created_at,
            'filepath': filepath,
            'filename': filename,
            'mimetype': file.mimetype or f'image/{ext}',
        })
        return redirect(url_for('view_paste', paste_id=paste_id))

    elif paste_type == 'file':
        file = request.files.get('file')
        if not file or file.filename == '':
            return render_template('index.html', expiry_options=EXPIRY_OPTIONS, error='No file selected'), 400

        filename = secure_filename(file.filename)
        paste_id = generate_id()

        if is_text_file(filename):
            content = file.read().decode('utf-8', errors='replace')
            save_text_content(paste_id, content)
            store_paste(paste_id, {
                'type': 'text',
                'title': title or filename,
                'expires_at': expires_at,
                'created_at': created_at,
                'filename': filename,
            })
        else:
            ext = filename.rsplit('.', 1)[-1] if '.' in filename else 'bin'
            saved_name = f"{paste_id}.{ext}"
            filepath = os.path.join(app.config['UPLOAD_FOLDER'], saved_name)
            file.save(filepath)

            store_paste(paste_id, {
                'type': 'file',
                'title': title or filename,
                'expires_at': expires_at,
                'created_at': created_at,
                'filepath': filepath,
                'filename': filename,
                'mimetype': file.mimetype or 'application/octet-stream',
            })
        return redirect(url_for('view_paste', paste_id=paste_id))

    else:
        return render_template('index.html', expiry_options=EXPIRY_OPTIONS, error='Invalid paste type'), 400


@app.route('/<paste_id>', methods=['GET'])
def view_paste(paste_id):
    paste = load_paste(paste_id)
    if not paste:
        abort(404)

    if is_expired(paste):
        delete_paste(paste_id)
        abort(404)

    if paste['type'] == 'text':
        paste['data'] = get_text_content(paste_id) or ''
        return render_template('view_text.html', paste=paste, paste_id=paste_id)
    elif paste['type'] == 'image':
        return render_template('view_image.html', paste=paste, paste_id=paste_id)
    elif paste['type'] == 'file':
        return render_template('view_file.html', paste=paste, paste_id=paste_id)


@app.route('/<paste_id>/download', methods=['GET'])
def download_file(paste_id):
    paste = load_paste(paste_id)
    if not paste:
        abort(404)

    if is_expired(paste):
        delete_paste(paste_id)
        abort(404)

    if paste['type'] == 'text':
        from io import BytesIO
        content = (get_text_content(paste_id) or '').encode('utf-8')
        return send_file(BytesIO(content), download_name=paste.get('filename', 'paste.txt'), as_attachment=True)
    elif paste['type'] in ('image', 'file'):
        filepath = paste.get('filepath')
        if not filepath or not os.path.exists(filepath):
            abort(404)
        return send_file(filepath, download_name=paste['filename'], as_attachment=True, mimetype=paste.get('mimetype', 'application/octet-stream'))


@app.route('/<paste_id>/raw', methods=['GET'])
def raw_paste(paste_id):
    paste = load_paste(paste_id)
    if not paste:
        abort(404)

    if is_expired(paste):
        delete_paste(paste_id)
        abort(404)

    if paste['type'] == 'text':
        content = get_text_content(paste_id) or ''
        return content, 200, {'Content-Type': 'text/plain; charset=utf-8'}
    elif paste['type'] == 'image':
        filepath = paste.get('filepath')
        if not filepath or not os.path.exists(filepath):
            abort(404)
        return send_file(filepath, mimetype=paste.get('mimetype', 'image/png'))


@app.errorhandler(413)
def request_entity_too_large(e):
    return render_template('index.html', expiry_options=EXPIRY_OPTIONS, error='File too large. Maximum size is 20 MB.'), 413


if __name__ == '__main__':
    ensure_dirs()
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))