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
| from flask import Flask, request, jsonify, g, Blueprint, Response, redirect
import uuid
from flask_caching import Cache
import os
import mariadb
import datetime
app = Flask(__name__)
# Configure caching (simple in-memory cache)
app.config["CACHE_TYPE"] = "RedisCache"
app.config["CACHE_REDIS_HOST"] = os.getenv("CACHE_REDIS_HOST", "redis")
app.config["CACHE_DEFAULT_TIMEOUT"] = 604800 # Cache expires in 7 days
cache = Cache(app)
def get_db_connection():
try:
conn = mariadb.connect(
host=os.getenv("DATABASE_HOST"),
user=os.getenv("DATABASE_USER"),
password=os.getenv("DATABASE_PASSWORD"),
database=os.getenv("DATABASE_NAME"),
)
return conn
except mariadb.Error as e:
return {"error": str(e)}
# I'm lazy to do this properly, so enjoy this ChatGPT'd run_query function!
def run_query(query, params=None):
conn = get_db_connection()
if isinstance(conn, dict):
return conn
try:
cursor = conn.cursor(dictionary=True)
cursor.execute(query, params or ())
conn.commit()
result = {
"success": True,
"affected_rows": cursor.rowcount,
"result": cursor.fetchall(),
}
return result
except mariadb.Error as e:
print("ERROR:", e, flush=True)
return {"error": str(e)}
finally:
cursor.close()
conn.close()
@app.route("/")
def index():
if "id" not in request.cookies:
unique_id = str(uuid.uuid4())
run_query("INSERT INTO users VALUES (%s, %s);", (unique_id, 0))
else:
unique_id = request.cookies.get("id")
res = run_query("SELECT * FROM users WHERE id = %s;", (unique_id,))
print(res, flush=True)
if "affected_rows" not in res:
print("ERRROR:", res)
return "ERROR"
if res["affected_rows"] == 0:
unique_id = str(uuid.uuid4())
run_query("INSERT INTO users VALUES (%s, %s);", (unique_id, 0))
html = f"""
<!DOCTYPE html>
<html>
<head>
<title>{unique_id}</title>
</head>
<body>
<h1>Your unique account ID: {unique_id}</h1>
<p><a href="/check?uuid={unique_id}">Click here to check if you are a winner!</a></p>
</body>
</html>
"""
r = Response(html)
r.set_cookie("id", unique_id)
return r
def normalize_uuid(uuid: str):
uuid_l = list(uuid)
i = 0
for i in range(len(uuid)):
uuid_l[i] = uuid_l[i].upper()
if uuid_l[i] == "-":
uuid_l.pop(i)
uuid_l.append(" ")
return "".join(uuid_l)
def make_cache_key():
return f"GET_check_uuids:{normalize_uuid(request.args.get('uuid'))}"[:64] # prevent spammers from filling redis cache
check_bp = Blueprint("check_bp", __name__)
@check_bp.route("/check")
@cache.cached(timeout=604800, make_cache_key=make_cache_key)
def check():
user_uuid = request.args.get("uuid")
if not user_uuid:
return {"error": "UUID parameter is required"}, 400
run_query("UPDATE users SET value = value + 1 WHERE id = %s;", (user_uuid,))
res = run_query("SELECT * FROM users WHERE id = %s;", (user_uuid,))
g.cache_hit = False
if "affected_rows" not in res:
print("ERRROR:", res)
return "Error"
if res["affected_rows"] == 0:
return "Invalid account ID"
num_wins = res["result"][0]["value"]
if num_wins >= 100:
return f"""CONGRATS! YOU HAVE WON.............. A FLAG! {os.getenv("FLAG")}"""
return f"""<p>Congrats! You have won! Only {100 - res["result"][0]["value"]} more wins to go.</p>
<p>Next attempt allowed at: {(datetime.datetime.now() + datetime.timedelta(days=7)).isoformat(sep=" ")} UTC</p><p><a href="/">Go back to the homepage</a></p>"""
# Hack to show to the user in the X-Cached header whether or not the response was cached
# How in the world does the flask caching library not support adding this header?????
@check_bp.after_request
def add_cache_header(response):
if hasattr(g, "cache_hit") and not g.cache_hit:
response.headers["X-Cached"] = "MISS"
else:
response.headers["X-Cached"] = "HIT"
g.cache_hit = True
return response
app.register_blueprint(check_bp)
# Debugging use for dev - remove before prod
# @app.route("/clear")
# def clear():
# cache.clear()
# return "cache cleared!"
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
|
