add protection that shares the data with my webside

my_modules/TheIPManager.py

@@ -0,0 +1,14 @@
+class TheIPManager:
+  def __init__(self):
+    self.always_allowed_ips = set()
+
+  # Checks
+  def is_client_ip_always_allowed(self, client_ip:str):
+    return client_ip in self.always_allowed_ips
+
+  # Add
+  def add_always_allowed_ip(self, client_ip:str):
+    self.always_allowed_ips.add(client_ip)
+
+  def update_always_allowed_ip(self, client_ips:set):
+    self.always_allowed_ips.update(client_ips)

my_modules/app/constens.py

@@ -1,7 +1,7 @@
+from my_modules.TheIPManager import TheIPManager
 from my_modules.app.logger import logger
 
 from dotenv import find_dotenv, load_dotenv, dotenv_values
-from pathlib import Path
 import os, asyncio
 
 async def read_dot_file():
@@ -17,5 +17,4 @@ WEB_DEBUG = os.getenv("WEB_DEBUG", False)
 SECRET_KEY = os.getenv("FLASK_SECRET_KEY", "USE_ENV_das_ist_ein_geheimer_schlüssel_1")
 API_GROUP = os.getenv("API_GROUP", 'NanoShare')
 
-UPLOAD_DIR = Path("uploads")
+THE_IP_BOT_MANAGER = TheIPManager()
-UPLOAD_DIR.mkdir(parents=True, exist_ok=True)

my_modules/app/logger.py

@@ -8,5 +8,5 @@ formatter = Formatter(fmt="%(levelname)s %(module)s: %(message)s")
 handler = AsyncStreamHandler(stream=sys.stdout)
 handler.formatter = formatter
 
-logger = Logger(name="my_webside_and_api", level="DEBUG" if os.getenv("WEB_DEBUG", False) == "true" else "INFO")
+logger = Logger(name="simple-picoshare", level="DEBUG" if os.getenv("WEB_DEBUG", False) == "true" else "INFO")
 logger.handlers = [handler]

my_modules/app/setup.py

@@ -1,8 +1,7 @@
-from my_modules.functions import custom_limit_key
+from my_modules.functions import custom_limit_key, get_my_ip_address, get_local_ip_addresses, replace_last_ip_segment, generate_all_ips
-from my_modules.app.constens import SECRET_KEY, UPLOAD_DIR
+from my_modules.app.constens import SECRET_KEY, THE_IP_BOT_MANAGER
 from my_modules.AsyncCache import AsyncCache
 from my_modules.app.logger import logger
-
 from my_modules.db.ConvexDB import ConvexDB
 
 from quart_session import Session
@@ -16,7 +15,6 @@ app = Quart(__name__, template_folder="../../templates/side", static_folder="../
 app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024 * 1024
 
 app.secret_key = SECRET_KEY
-app.upload_folder = UPLOAD_DIR
 
 # Cache, Sessions and Limiter over Valkey
 if os.getenv("VALKEY_HOST", None) is not None:
@@ -71,6 +69,15 @@ async def init_convex():
   app.convex = ConvexDB(os.getenv("CONVEX_URL"))
   await app.convex.connect()
 
+  THE_IP_BOT_MANAGER.add_always_allowed_ip('127.0.0.1')
+  THE_IP_BOT_MANAGER.add_always_allowed_ip(await get_my_ip_address())
+
+  local_docker_host_ip = get_local_ip_addresses()
+  if local_docker_host_ip:
+    base_ip = replace_last_ip_segment(local_docker_host_ip, 1)
+    all_local_ips = generate_all_ips(base_ip)
+    THE_IP_BOT_MANAGER.update_always_allowed_ip(all_local_ips)
+
 @app.after_serving
 async def close_convex():
   if app.convex:

my_modules/decoratory/header.py

@@ -1,4 +1,4 @@
-from my_modules.app.constens import SECRET_KEY
+from my_modules.app.constens import THE_IP_BOT_MANAGER
 from my_modules.app.logger import logger
 from my_modules.app.setup import LIMITER
 from my_modules.functions import get_ip
@@ -6,7 +6,7 @@ from my_modules.functions import get_ip
 from quart import jsonify, request, url_for, Response, current_app, session, abort
 from functools import wraps
 from datetime import datetime
-import asyncio, msgpack, json, jwt
+import asyncio, msgpack, json
 
 def encode_object_default(obj):
   if isinstance(obj, datetime):
@@ -24,6 +24,27 @@ async def get_auth_token():
 
   return None
 
+async def verify_token(token:str):
+  decoded_payload = await current_app.convex.decode_access_token_payload(access_token=token)
+  decoded_payload_error_state = decoded_payload.get('state', None)
+
+  if decoded_payload is None:
+    return {'error': "No Data from Database"}, 504
+  elif decoded_payload_error_state == 1:
+    await logger.error(decoded_payload.get('error'))
+    return {'error': 'Invalid access token'}, 401
+  elif decoded_payload_error_state == 2:
+    await logger.error(decoded_payload.get('error'))
+    return {'error': 'Wrong access token type'}, 401
+  elif decoded_payload_error_state == 3:
+    await logger.error(decoded_payload.get('error'))
+    return {'error': 'Refresh token not found', 'msg': 'Please login again and generate a new Token', 'url': url_for('auth_login.login')}, 403
+  elif decoded_payload_error_state == 4:
+    await logger.error(decoded_payload.get('error'))
+    return {'error': 'Refresh token expired'}, 401
+
+  return decoded_payload, None
+
 # Custom decorator for token validation
 def token_required(func):
   @wraps(func)
@@ -33,17 +54,10 @@ def token_required(func):
       await logger.error('API Token is missing')
       return jsonify(error='Token is missing'), 400
 
-    try:
+    decoded_payload, status_code = await verify_token(token)
-      decoded_payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
+    decoded_payload_error = decoded_payload.get('error', None)
-      if not await current_app.edgedb.check_if_refresh_token_exists_by_id(decoded_payload['refresh_id']):
+    if decoded_payload_error:
-        await logger.error(f'API Refresh Token not found: {decoded_payload['refresh_id']}')
+      return jsonify(decoded_payload), status_code
-        return jsonify(error='Refresh Token not found', msg='Please login again', url=url_for('login')), 403
-    except jwt.ExpiredSignatureError:
-      await logger.error('API Token has expired')
-      return jsonify(error='Token has expired'), 401
-    except jwt.InvalidTokenError:
-      await logger.error('API Token is invalid')
-      return jsonify(error='Token is invalid'), 401
 
     return await func(user=decoded_payload, *args, **kwargs)
   return wrapper
@@ -116,8 +130,8 @@ def apply_limit(endpoint_name, limits:dict=None):
   def make_key_func(endpoint):
     def key_func():
       ip = get_ip()
-      # if THE_IP_BOT_MANAGER.is_client_ip_always_allowed(ip):
+      if THE_IP_BOT_MANAGER.is_client_ip_always_allowed(ip):
-      #   return None  # No key, no increment, no enforcement
+        return None  # No key, no increment, no enforcement
 
       # Combine endpoint name and HTTP method (and client IP) into the rate-limit key
       return f":{ip}:{endpoint}:{request.method}:"

my_modules/middleware.py

@@ -1,3 +1,6 @@
+from routes.handeling.errorsAndBots import maybe_a_hacker
+
+from my_modules.app.constens import THE_IP_BOT_MANAGER
 from my_modules.app.logger import logger
 from my_modules.functions import get_ip
 from my_modules.app.setup import app
@@ -14,13 +17,29 @@ async def custom_middleware():
   path = request.path
   method = request.method
 
+  db_whitelisted_or_blocked = await current_app.convex.is_ip_address_whitelisted_or_blocked(ip_address=client_ip)
+
   # Skip allowed IPs or non-critical assets
   if (
-    "favicon" in path
+    db_whitelisted_or_blocked['whiteliste']
+    or THE_IP_BOT_MANAGER.is_client_ip_always_allowed(client_ip)
     or "static" in path
+    or "favicon" in path
+    or "storage" in path
   ):
     return
 
+  # 2. If IP is already blocked
+  if db_whitelisted_or_blocked['blocked']:
+    await logger.error(f"[BLOCKED] {method} | {client_ip} tried {method} {path}")
+    await current_app.convex.increment_blocked_ip_address_access(ip_address=client_ip, method=method, path=path)
+    return await render_template("views/basics/blocked_access.htm", remote_addr=client_ip), 403
+
+  # 3. If path contains honeypot targets
+  if await current_app.convex.is_path_blocked(path=path):
+    await logger.warning(f"[HONEYPOT] {method} | {client_ip} accessed {path}")
+    return await maybe_a_hacker()
+
   await logger.info(f"{method} | {client_ip} had accessed the Side {path}")
 
 @app.context_processor

routes/handeling/errorsAndBots.py

@@ -1,18 +1,20 @@
 from my_modules.app.setup import app, LIMITER
 from my_modules.app.logger import logger
 
-from quart import request, render_template, jsonify, current_app, make_response, redirect, url_for
+from quart import request, render_template, jsonify, current_app, make_response
 from my_modules.functions import get_ip, enforce_custom_limit
 
 @app.errorhandler(401)
 async def handle_unauthorized(e):
-  try:
+  if request.path.startswith("/api"):
-    enforce_custom_limit(LIMITER, "401", limit_count=5, window_sec=1800)
+    return jsonify({"error": "Unauthorized Access", "message": "Gandalf has spoken: You shall not pass… until you log in."}), 401
-  except LookupError as e:
-    return await to_many_requests(e)
 
   await logger.error(e)
-  return redirect(url_for('auth_login.login'))
+  return await render_template('views/basics/error.htm',
+    title='Unauthorized Access',
+    header={'title': '401 - Unauthorized', 'message': "Gandalf has spoken: You shall not pass… until you log in."},
+    file={'name': '401.gif', 'alt': "Gandalf blocking the bridge – You shall not pass!"},
+  ), 401
 
 @app.errorhandler(404)
 async def not_found(e):
@@ -21,13 +23,44 @@ async def not_found(e):
   except LookupError as e:
     return await to_many_requests(e)
 
+  if request.path.startswith("/api"):
+    return jsonify({"error": "Page Not Found", "message": "Oops! The page you are looking for does not exist."}), 404
+
   await logger.error(f"[404] Page Not Found: {request.path}")
+  await current_app.convex.increment_page_not_found_error(path=request.path, status=404)
+
   return await render_template('views/basics/error.htm',
     title='Page Not Found',
     header={'title': '404 - Page Not Found', 'message': "Oops! The page you are looking for does not exist."},
     file={'name': '404.webp', 'alt': "Matrix - Neo stoping the Bullets by holding his hand up"},
   ), 404
 
+@app.errorhandler(418)
+async def maybe_a_hacker(e=None):
+  try:
+    enforce_custom_limit(LIMITER, "BotScan", 5, 120)
+  except LookupError as e:
+    client_ip=get_ip()
+    await current_app.convex.increment_blocked_ip_address_access(
+      ip_address=client_ip,
+      method=request.method,
+      path=request.path,
+    )
+    await logger.warning(f"[HONEYPOT] Blocked {client_ip} after accessing {request.path}")
+    return await to_many_requests(e)
+
+  rendered = await render_template('views/basics/error.htm',
+    title='Oops! Something Went AWOL!',
+    header={'title': "418 - I'm a Teapot", 'message': f"You don't say the Magic Word. By the way, we might have your IP now, but don’t worry, it's in safe hands (probably). Feel free to keep poking around, just maybe give us a sec to catch our breath."},
+    file={'name': 'hacker_crap.webp', 'alt': "Someone got Hacked and he says I hate this Hacker crap - Jurassic Park Movie"},
+  )
+
+  response = await make_response((rendered, 418))
+  response.headers['X-Honeypot-Triggered'] = 'true'
+  response.headers['X-Reason'] = 'Unauthorized access attempt'
+
+  return response
+
 @app.errorhandler(429)
 async def to_many_requests(e):
   message = "We love your enthusiasm, but our server thought it was being DDoSed… by you. The keyboard needs a new set of keys and we need a nap. Try again soon!"
@@ -48,6 +81,9 @@ async def internal_server_error(e):
   except LookupError as e:
     return await to_many_requests(e)
 
+  if request.path.startswith("/api"):
+    return jsonify({"error": "Internal Server Error", "message": "It looks like you broke something... but don't worry, we're fixing it! In the meantime, we may or may not have logged your IP address (just kidding... or are we?). Either way, thanks for helping us find new ways to crash our system. Stay curious, hacker-friend!"}), 500
+
   await logger.error(e)
   return await render_template('views/basics/error.htm',
     title='Internal Server Error',

templates/static/css/blocked_access.css

@@ -0,0 +1,37 @@
+body {
+  font-family: Arial, sans-serif;
+  background-color: var(--primary-color);
+  margin: 0;
+  padding: 0;
+  display: flex;
+  justify-content: center;
+  align-items: center;
+  height: 100vh;
+  text-align: center;
+  color: var(--text-color);
+}
+
+.container {
+  background-color: var(--secondary-color);
+  padding: 40px;
+  border-radius: 10px;
+  box-shadow: 0 4px 8px rgba(0, 0, 0, 0.1);
+  max-width: 920px;
+}
+
+h1 {
+  color: #e74c3c;
+  font-size: 48px;
+  margin: 0;
+  font-family: 'Press Start 2P', cursive;
+}
+
+p {
+  font-size: 18px;
+  margin-top: 10px;
+}
+
+.blocked-ip {
+  font-weight: bold;
+  color: var(--text-muted);
+}

templates/static/css/root.css

@@ -0,0 +1,55 @@
+:root {
+  --primary-color: #f0f0f0;
+  --secondary-color: white;
+
+  --nav-background-color: #333;
+
+  --text-color: #000;
+  --text-muted: #555;
+
+  --table-head-background-color: #e1e1e1;
+
+  --container-width: 580px;
+  --container-max-width: 903px;
+
+  --furry-button-color: #ffcaa6;
+  --furry-button-color-hover: #ffa974;
+
+  --furry-background-color: white;
+  --furry-decoration-color: #ff7f50;
+
+  --furry-list-element-color: #fff4e6;
+  --furry-list-element-color-hover: #ffeede;
+
+  --story-text-color: #33ff33;
+  --story-press-any-key-color: #ccc;
+  --story-background-color: radial-gradient(circle at center, #222 0%, #000 100%);
+
+  --story-button-color: #0d0d0d;
+  --story-button-hover-color: #003300;
+
+  --story-sidebar-background-color: black;
+  --story-boot-logo-color: #ffaa00;
+}
+
+@media (prefers-color-scheme: dark) {
+  :root {
+    /* --primary-color: #353535;
+    --secondary-color: #1f1f23; */
+    /* --primary-color: #2a2640;
+    --secondary-color: #3a3458; */
+    --primary-color: #1e1e2f;
+    --secondary-color: #2a2a3d;
+
+    --nav-background-color: #272424;
+
+    --text-color: #cec9da;
+    --text-muted: #999;
+
+    --table-head-background-color: var(--secondary-color);
+
+    --furry-button-color: #b67616;
+    --furry-list-element-color: #252231;
+    --furry-list-element-color-hover: var(--primary-color);
+  }
+}