-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathzluqet.py
252 lines (208 loc) · 7.79 KB
/
zluqet.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
from flask import Flask, render_template, request, redirect, url_for, abort, make_response, jsonify
from flask_sqlalchemy import SQLAlchemy
from pygments import highlight
from pygments.lexers import guess_lexer, get_lexer_by_name
from pygments.formatters import HtmlFormatter
from pygments.util import ClassNotFound
import string
import random
from datetime import datetime
import time
from collections import defaultdict, deque
from sqlalchemy.engine import Engine
from sqlalchemy import event
from types import SimpleNamespace
from cachetools import LRUCache
from werkzeug.exceptions import RequestEntityTooLarge
from werkzeug.middleware.proxy_fix import ProxyFix
app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_host=1)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///pastes.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_EXPIRE_ON_COMMIT'] = False
app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
'pool_size': 10,
'max_overflow': 20,
'pool_pre_ping': True,
}
app.secret_key = 'zluqet'
db = SQLAlchemy(app)
@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA synchronous = OFF")
cursor.execute("PRAGMA journal_mode = WAL")
cursor.execute("PRAGMA cache_size = 10000")
cursor.close()
ip_requests = defaultdict(deque)
EXEMPT_IPS = {}
REQUESTS_PER_MINUTE = 5
@app.before_request
def rate_limit():
if request.path.startswith('/api/'):
client_ip = request.remote_addr
if client_ip in EXEMPT_IPS:
return
now = time.time()
timestamps = ip_requests[client_ip]
while timestamps and now - timestamps[0] > 60:
timestamps.popleft()
if len(timestamps) >= REQUESTS_PER_MINUTE:
return jsonify({'error': 'Too many requests, please slow down.'}), 429
timestamps.append(now)
class Paste(db.Model):
id = db.Column(db.Integer, primary_key=True)
content = db.Column(db.Text, nullable=False)
key = db.Column(db.String(8), unique=True, nullable=False)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
def __repr__(self):
return f'<Paste {self.key}>'
# Use LRU caches to limit memory usage
paste_cache = LRUCache(maxsize=1000)
highlight_cache = LRUCache(maxsize=1000)
def cache_paste(p):
return SimpleNamespace(
content=p.content,
key=p.key,
created_at=p.created_at
)
def get_paste_by_key(key):
if key in paste_cache:
return paste_cache[key]
paste = Paste.query.filter_by(key=key).first()
if paste:
lite = cache_paste(paste)
paste_cache[key] = lite
return lite
return None
formatter = HtmlFormatter(style="default")
STYLE_DEFS = formatter.get_style_defs('.highlight')
def generate_key(length=8):
chars = string.ascii_uppercase + string.digits
while True:
key = ''.join(random.choices(chars, k=length))
if not Paste.query.filter_by(key=key).first():
return key
def precompute_highlighting(content, key):
try:
lexer = guess_lexer(content)
except ClassNotFound:
lexer = get_lexer_by_name("text", stripall=True)
highlighted = highlight(content, lexer, formatter)
highlight_cache[key] = highlighted
MAX_LENGTH = 25000
@app.route('/', methods=['GET', 'POST'])
def index():
if request.method == 'POST':
content = request.form.get('content')
if not content:
return redirect(url_for('index'))
if len(content) > MAX_LENGTH:
error_message = (
f"Your paste exceeds the maximum allowed character limit "
f"of {MAX_LENGTH} characters. Please reduce your content before saving."
)
return render_template('index.html', error=error_message, content=content)
key = generate_key()
new_paste = Paste(content=content, key=key)
db.session.add(new_paste)
db.session.commit()
lite = cache_paste(new_paste)
paste_cache[key] = lite
precompute_highlighting(new_paste.content, key)
return redirect(url_for('view_paste', key=key))
return render_template('index.html')
@app.route('/<key>')
def view_paste(key):
paste = get_paste_by_key(key)
if not paste:
abort(404)
highlighted_code = highlight_cache.get(key)
if not highlighted_code:
try:
lexer = guess_lexer(paste.content)
except ClassNotFound:
lexer = get_lexer_by_name("text", stripall=True)
highlighted_code = highlight(paste.content, lexer, formatter)
highlight_cache[key] = highlighted_code
response = make_response(
render_template(
'view_paste.html',
paste=paste,
highlighted_code=highlighted_code,
style_defs=STYLE_DEFS
)
)
response.headers['X-Robots-Tag'] = 'noindex, nofollow'
return response
@app.route('/raw/<key>')
def raw_paste(key):
paste = get_paste_by_key(key)
if not paste:
abort(404)
return paste.content, {'Content-Type': 'text/plain'}
@app.route('/edit/<key>', methods=['GET', 'POST'])
def dupe_paste(key):
paste = get_paste_by_key(key)
if not paste:
abort(404)
if request.method == 'POST':
content = request.form.get('content')
if not content:
return redirect(url_for('index'))
if len(content) > MAX_LENGTH:
error_message = (
f"Your paste exceeds the maximum allowed character limit "
f"of {MAX_LENGTH} characters. Please reduce your content before saving."
)
return render_template('edit_paste.html', error=error_message, paste=paste, content=content)
new_key = generate_key()
new_paste = Paste(content=content, key=new_key)
db.session.add(new_paste)
db.session.commit()
lite = cache_paste(new_paste)
paste_cache[new_key] = lite
precompute_highlighting(new_paste.content, new_key)
return redirect(url_for('view_paste', key=new_key))
return render_template('edit_paste.html', paste=paste)
@app.route('/api/documents', methods=['POST'])
def api_create_paste():
text_content = request.get_data(as_text=True)
if not text_content:
return jsonify({'error': 'No content provided.'}), 400
if len(text_content) > MAX_LENGTH:
return jsonify({'error': f"Your paste exceeds the maximum allowed character limit of {MAX_LENGTH} characters."}), 400
key = generate_key()
new_paste = Paste(content=text_content, key=key)
db.session.add(new_paste)
db.session.commit()
lite = cache_paste(new_paste)
paste_cache[key] = lite
precompute_highlighting(new_paste.content, key)
response_data = {'key': key}
return jsonify(response_data), 200
@app.route('/api/documents/<key>', methods=['GET'])
def api_get_paste(key):
paste = get_paste_by_key(key)
if not paste:
return jsonify({'error': 'Paste not found.'}), 404
return jsonify({
'key': paste.key,
'content': paste.content,
'created_at': paste.created_at.isoformat()
})
@app.errorhandler(404)
def page_not_found(e):
return render_template('errors/404.html'), 404
@app.errorhandler(RequestEntityTooLarge)
def handle_request_entity_too_large(error):
error_message = (
f"Your paste is too large! The maximum allowed size is {MAX_LENGTH} characters. "
"Please reduce your content and try again."
)
return render_template("errors/too_large.html", error=error_message), 413
if __name__ == '__main__':
with app.app_context():
db.create_all()
app.run(debug=False, threaded=True)