diff --git a/app.py b/app.py index 54574ee..fecd9fa 100644 --- a/app.py +++ b/app.py @@ -13,6 +13,7 @@ from lightdb import LightDB from tools.geo_loc import geo_loc_bp from tools.auth import auth_bp +from tools.db_auth import is_admin_user from werkzeug.exceptions import RequestEntityTooLarge from werkzeug.utils import secure_filename @@ -78,7 +79,8 @@ def dashboard(): files=userfiles, username=username, quota_gb=udb[[user['username'] for user in udb].index(username)]['quota_gb'], - quota_usage=quota_usage_gb + quota_usage=quota_usage_gb, + is_admin=is_admin_user(username) ) @app.route('/upload') @@ -307,5 +309,141 @@ def backRename(): flash('Invalid code! Check if you typed the correct code, and for one-time codes, make sure nobody else entered the code before you did.', 'error') return redirect(url_for('upload')) +@app.route('/admin') +def admin(): + if not session.get('loggedIn', False): + flash('You must be signed in to access the admin panel!', 'error') + return redirect(url_for('upload')) + + username = session.get('username') + if not is_admin_user(username): + flash('Access denied. Admin privileges required.', 'error') + return redirect(url_for('upload')) + + db = l_db['files'] + udb = l_db['users'] + + # Calculate per-user storage usage + user_usage = {} + user_file_count = {} + total_storage_mb = 0.0 + + for file_key in db: + owner = db[file_key].get('owner') + size_mb = db[file_key].get('size_megabytes', 0) or 0 + total_storage_mb += size_mb + if owner: + user_usage[owner] = user_usage.get(owner, 0) + size_mb + user_file_count[owner] = user_file_count.get(owner, 0) + 1 + + total_storage_gb = round(total_storage_mb / 1024, 2) + total_files = len(db) + + users_data = [] + for user in udb: + uname = user.get('username', '') + usage_mb = user_usage.get(uname, 0) + usage_gb = round(usage_mb / 1024, 2) + users_data.append({ + 'username': uname, + 'quota_gb': user.get('quota_gb', 0), + 'usage_gb': usage_gb, + 'file_count': user_file_count.get(uname, 0), + 'is_admin': user.get('is_admin', False) + }) + + return render_template( + 'admin.html', + username=username, + users=users_data, + total_storage_gb=total_storage_gb, + total_files=total_files, + total_users=len(users_data) + ) + + +@app.route('/admin/update_quota', methods=['POST']) +def admin_update_quota(): + if not session.get('loggedIn', False) or not is_admin_user(session.get('username')): + flash('Access denied.', 'error') + return redirect(url_for('upload')) + + target = request.form.get('username') + try: + new_quota = float(request.form.get('quota_gb', 0)) + except (ValueError, TypeError): + flash('Invalid quota value.', 'error') + return redirect(url_for('admin')) + + udb = l_db['users'] + users = list(udb) + for i, user in enumerate(users): + if user.get('username') == target: + users[i]['quota_gb'] = new_quota + l_db['users'] = users + flash(f"Quota for {target} updated to {new_quota}GB.", 'info') + return redirect(url_for('admin')) + + flash('User not found.', 'error') + return redirect(url_for('admin')) + + +@app.route('/admin/delete_user', methods=['POST']) +def admin_delete_user(): + if not session.get('loggedIn', False) or not is_admin_user(session.get('username')): + flash('Access denied.', 'error') + return redirect(url_for('upload')) + + target = request.form.get('username') + if target == session.get('username'): + flash('You cannot delete your own account.', 'error') + return redirect(url_for('admin')) + + udb = l_db['users'] + users = list(udb) + new_users = [u for u in users if u.get('username') != target] + if len(new_users) == len(users): + flash('User not found.', 'error') + return redirect(url_for('admin')) + + l_db['users'] = new_users + # Remove files owned by the deleted user + db = l_db['files'] + files_to_delete = [fkey for fkey in db if db[fkey].get('owner') == target] + for fkey in files_to_delete: + try: + os.remove(os.path.join(app.config['UPLOAD_DIRECTORY'], fkey)) + except FileNotFoundError: + pass + db.pop(fkey) + flash(f"User {target} has been deleted.", 'info') + return redirect(url_for('admin')) + + +@app.route('/admin/toggle_admin', methods=['POST']) +def admin_toggle_admin(): + if not session.get('loggedIn', False) or not is_admin_user(session.get('username')): + flash('Access denied.', 'error') + return redirect(url_for('upload')) + + target = request.form.get('username') + if target == session.get('username'): + flash('You cannot change your own admin status.', 'error') + return redirect(url_for('admin')) + + udb = l_db['users'] + users = list(udb) + for i, user in enumerate(users): + if user.get('username') == target: + users[i]['is_admin'] = not bool(user.get('is_admin', False)) + l_db['users'] = users + status = 'granted' if users[i]['is_admin'] else 'revoked' + flash(f"Admin privileges {status} for {target}.", 'info') + return redirect(url_for('admin')) + + flash('User not found.', 'error') + return redirect(url_for('admin')) + + port = int(os.environ.get("ADRIVE_PORT", 3133)) app.run(debug=True, port=port, host='0.0.0.0') diff --git a/templates/admin.html b/templates/admin.html new file mode 100644 index 0000000..e6eb2e1 --- /dev/null +++ b/templates/admin.html @@ -0,0 +1,188 @@ + + + + + + + + ADrive Admin Panel + + + + + + + +
+ {% with messages = get_flashed_messages(with_categories=true) %} + {% if messages %} + + {% endif %} + {% endwith %} + +
+
+
+

Admin Panel

+

Signed in as {{ username }}

+
+
+ + +
+
+
+
+
Total Users
+

{{ total_users }}

+
+
+
+
+
+
+
Total Files
+

{{ total_files }}

+
+
+
+
+
+
+
Total Storage Used
+

{{ total_storage_gb }} GB

+
+
+
+
+ + +

User Management

+
+ + + + + + + + + + + + + + {% for user in users %} + + + + + + + + + + {% else %} + + + + {% endfor %} + +
UsernameAdminFilesStorage UsedQuotaUsageActions
+ {{ user.username }} + {% if user.username == username %} + you + {% endif %} + + {% if user.is_admin %} + Admin + {% else %} + User + {% endif %} + {{ user.file_count }}{{ user.usage_gb }} GB{{ user.quota_gb }} GB + {% if user.quota_gb and user.quota_gb > 0 %} + {% set pct = ((user.usage_gb / user.quota_gb) * 100) | round | int %} + {% else %} + {% set pct = 0 %} + {% endif %} +
+
{{ pct }}%
+
+ 		+
+ +
+ + + + +
+ {% if user.username != username %} + +
+ + +
+ +
+ + +
+ {% endif %} +
+
No users found.
+
+ +
+ + + +
+
+
+ + + + + + diff --git a/templates/dashboard.html b/templates/dashboard.html index 63e02b3..445e205 100644 --- a/templates/dashboard.html +++ b/templates/dashboard.html @@ -113,6 +113,10 @@

Dashboard of {{username}}

Upload a New File + {% if is_admin %} + + Admin Panel + {% endif %} Sign Out diff --git a/tools/db_auth.py b/tools/db_auth.py index 8f01e3c..edd6938 100644 --- a/tools/db_auth.py +++ b/tools/db_auth.py @@ -16,18 +16,36 @@ def _check_password(password: str, hashed_password: str) -> bool: return bcrypt.checkpw(password.encode('utf-8'), hashed_password.encode('utf-8')) -def register_user(username: str, password: str, quota_gb: int = 5) -> None: +def register_user(username: str, password: str, quota_gb: int = 5, is_admin: bool = False) -> None: global l_db + if 'users' not in l_db: + l_db['users'] = [] + if 'files' not in l_db: + l_db['files'] = {} + + users = l_db['users'] + for i, user in enumerate(users): + if user.get('username') == username: + updated = dict(user) + updated['password'] = _hash_password(password) + updated['quota_gb'] = quota_gb + updated['is_admin'] = is_admin + users[i] = updated + return + user = { 'username': username, 'password': _hash_password(password), - 'quota_gb': quota_gb + 'quota_gb': quota_gb, + 'is_admin': is_admin } l_db['users'].append(user) def check_if_user_exists(username: str) -> bool: global l_db + if 'users' not in l_db: + return False for user in l_db['users']: if user.get('username') == username: return True @@ -37,8 +55,19 @@ def list_users() -> list: global l_db return l_db['users'] +def is_admin_user(username: str) -> bool: + global l_db + if 'users' not in l_db: + return False + for user in l_db['users']: + if user.get('username') == username: + return bool(user.get('is_admin', False)) + return False + def check_if_auth(username: str, password: str) -> bool: global l_db + if 'users' not in l_db: + return False for user in l_db['users']: if user.get('username') != username: continue diff --git a/update_database.py b/update_database.py index ff3b83f..455c97f 100644 --- a/update_database.py +++ b/update_database.py @@ -1,5 +1,3 @@ -from lightdb import LightDB +from tools.db_auth import register_user -l_db = LightDB() -l_db['users'] = [] -l_db['files'] = {} +register_user('admin', 'admin', quota_gb=10, is_admin=True)