Implement session authentication (removing uuid from url)

hiddifypanel/base.py

@@ -1,30 +1,33 @@
 import flask_bootstrap
 import hiddifypanel
-from dynaconf import FlaskDynaconf
-from flask import Flask, request, g
+from flask import request, g
 from flask_babelex import Babel
-from hiddifypanel.panel.init_db import init_db
+from flask_session import Session
+from flask_cors import CORS
+import datetime
 
-from hiddifypanel.models import *
 from dotenv import dotenv_values
 import os
 from hiddifypanel.panel import hiddify
 from apiflask import APIFlask
 from werkzeug.middleware.proxy_fix import ProxyFix
+from hiddifypanel.models import *
+from hiddifypanel.panel.init_db import init_db
+from hiddifypanel.cache import redis_client
 
 
 def create_app(cli=False, **config):
 
     app = APIFlask(__name__, static_url_path="/<proxy_path>/static/", instance_relative_config=True, version='2.0.0', title="Hiddify API",
-                   openapi_blueprint_url_prefix="/<proxy_path>/<user_secret>/api", docs_ui='elements', json_errors=False, enable_openapi=True)
+                   openapi_blueprint_url_prefix="/<proxy_path>/api", docs_ui='elements', json_errors=False, enable_openapi=True)
     # app = Flask(__name__, static_url_path="/<proxy_path>/static/", instance_relative_config=True)
     app.wsgi_app = ProxyFix(
         app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1
     )
     app.servers = {
         'name': 'current',
         'url': '',
-    }
+    }  # type: ignore
     app.info = {
         'description': 'Hiddify is a free and open source software. It is as it is.',
         'termsOfService': 'http://hiddify.com',
@@ -47,6 +50,15 @@ def create_app(cli=False, **config):
 
         app.config[c] = v
 
+    # setup flask server-side session
+    # app.config['APPLICATION_ROOT'] = './'
+    # app.config['SESSION_COOKIE_DOMAIN'] = '/'
+    app.config['SESSION_TYPE'] = 'redis'
+    app.config['SESSION_REDIS'] = redis_client
+    app.config['SESSION_PERMANENT'] = False
+    app.config['PERMANENT_SESSION_LIFETIME'] = datetime.timedelta(days=7)
+    Session(app)
+
     app.jinja_env.line_statement_prefix = '%'
     app.jinja_env.filters['b64encode'] = hiddify.do_base_64
     app.view_functions['admin.static'] = {}  # fix bug in apiflask
@@ -57,6 +69,7 @@ def create_app(cli=False, **config):
     with app.app_context():
         init_db()
 
+    hiddifypanel.panel.auth.init_app(app)
     hiddifypanel.panel.common.init_app(app)
     hiddifypanel.panel.admin.init_app(app)
     hiddifypanel.panel.user.init_app(app)
@@ -99,7 +112,10 @@ def check_csrf():
     @app.after_request
     def apply_no_robot(response):
         response.headers["X-Robots-Tag"] = "noindex, nofollow"
+        if response.status_code == 401:
+            response.headers['WWW-Authenticate'] = 'Basic realm="Hiddify"'
         return response
+
     app.jinja_env.globals['get_locale'] = get_locale
     app.jinja_env.globals['version'] = hiddifypanel.__version__
     app.jinja_env.globals['static_url_for'] = hiddify.static_url_for

hiddifypanel/hutils/utils.py

@@ -1,32 +1,17 @@
-import socket
+import base64
+from typing import Any, Tuple
+from urllib.parse import urlparse
 from uuid import UUID
-import user_agents
-from sqlalchemy.orm import Load
-import glob
-import json
-from babel.dates import format_timedelta as babel_format_timedelta
-from flask_babelex import gettext as __
 from flask_babelex import lazy_gettext as _
-import datetime
-from flask import jsonify, g, url_for, Markup, abort, current_app, request
+from datetime import datetime
+from flask import url_for, Markup
 from flask import flash as flask_flash
 import re
-from wtforms.validators import ValidationError
 import requests
 
 import string
 import random
-from babel.dates import format_timedelta as babel_format_timedelta
-import urllib
-import time
 import os
-import psutil
-from urllib.parse import urlparse
-import ssl
-import h2.connection
-import subprocess
-import netifaces
-import time
 import sys
 
 from hiddifypanel.cache import cache
@@ -37,13 +22,15 @@ def url_encode(strr):
     import urllib.parse
     return urllib.parse.quote(strr)
 
-def is_uuid_valid(uuid,version):
+
+def is_uuid_valid(uuid, version):
     try:
         uuid_obj = UUID(uuid, version=version)
     except ValueError:
         return False
     return str(uuid_obj) == uuid
 
+
 def error(str):
 
     print(str, file=sys.stderr)
@@ -53,11 +40,14 @@ def static_url_for(**values):
     orig = url_for("static", **values)
     return orig.split("user_secret")[0]
 
+
 @cache.cache(ttl=60000)
 def get_latest_release_url(repo):
-            latest_url = requests.get(f'{repo}/releases/latest').url.strip()
-            version = latest_url.split('tag/')[1].strip()
-            return (latest_url,version)
+    latest_url = requests.get(f'{repo}/releases/latest').url.strip()
+    version = latest_url.split('tag/')[1].strip()
+    return (latest_url, version)
+
+
 @cache.cache(ttl=600)
 def get_latest_release_version(repo_name):
     try:
@@ -104,13 +94,29 @@ def get_random_string(min_=10, max_=30):
     return result_str
 
 
+def get_random_password(length: int = 16) -> str:
+    '''Retunrns a random password with fixed length'''
+    characters = string.ascii_letters + string.digits  # + '-'
+    while True:
+        passwd = ''.join(random.choice(characters) for i in range(length))
+        if (any(c.islower() for c in passwd) and any(c.isupper() for c in passwd) and sum(c.isdigit() for c in passwd) > 1):
+            return passwd
+
+
+def is_assci_alphanumeric(str):
+    for c in str:
+        if c not in string.ascii_letters + string.digits:
+            return False
+    return True
+
+
 def date_to_json(d):
     return d.strftime("%Y-%m-%d") if d else None
 
 
 def json_to_date(date_str):
     try:
-        return datetime.datetime.strptime(date_str, '%Y-%m-%d')
+        return datetime.strptime(date_str, '%Y-%m-%d')
     except:
         return date_str
 
@@ -122,11 +128,97 @@ def time_to_json(d):
 
 def json_to_time(time_str):
     try:
-        return datetime.datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
+        return datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
     except:
         return time_str
 
 
 def flash(message, category):
     print(message)
     return flask_flash(Markup(message), category)
+
+
+def get_proxy_path_from_url(url: str) -> str | None:
+    url_path = urlparse(url).path
+    proxy_path = url_path.lstrip('/').split('/')[0] or None
+    return proxy_path
+
+
+def is_uuid_in_url_path(path: str) -> bool:
+    for section in path.split('/'):
+        if is_uuid_valid(section, 4):
+            return True
+    return False
+
+
+def get_uuid_from_url_path(path: str, section_index: int = 2) -> str | None:
+    """
+    Takes a URL path and extracts the UUID at the specified section index.
+
+    Args:
+        path (str): The URL path from which to extract the UUID.
+        section_index (int, optional): The index of the section in the URL path where the UUID is located. Defaults to 2, because in past the UUID was in the second section of path of url.
+
+    Returns:
+        str | None: The extracted UUID as a string if found, or None if not found.
+    """
+    s_index = 1
+    for section in path.lstrip('/').split('/'):
+        if is_uuid_valid(section, 4):
+            if s_index == section_index:
+                return section
+        s_index += 1
+    return None
+
+
+def get_apikey_from_auth_header(auth_header: str) -> str | None:
+    if auth_header.startswith('ApiKey'):
+        return auth_header.split('ApiKey ')[1].strip()
+    return None
+
+
+def get_basic_auth_from_auth_header(auth_header: str) -> str | None:
+    if auth_header.startswith('Basic'):
+        return auth_header.split('Basic ')[1].strip()
+    return None
+
+
+def parse_basic_auth_header(auth_header: str) -> tuple[str, str] | None:
+    if not auth_header.startswith('Basic'):
+        return None
+    header_value = auth_header.split('Basic ')
+    if len(header_value) < 2:
+        return None
+    username, password = map(lambda item: item.strip(), base64.urlsafe_b64decode(header_value[1].strip()).decode('utf-8').split(':'))
+    return (username, password) if username and password else None
+
+
+def parse_login_id(raw_id) -> Tuple[Any | None, str | None]:
+    """
+    Parses the given raw ID to extract the account type and ID.
+    Args:
+        raw_id (str): The raw ID to be parsed.
+    Returns:
+        Tuple[Any | None, str | None]: A tuple containing the account type and ID.
+            The account type is of type Role (either Role.user or Role.admin)
+            and the ID is a string. If the raw ID cannot be parsed, None is returned
+            for both the account type and ID.
+    """
+    splitted = raw_id.split('_')
+    if len(splitted) < 2:
+        return None, None
+    admin_or_user, id = splitted
+    from hiddifypanel.models.role import Role
+    account_type = Role.admin if admin_or_user == 'admin' else Role.user
+    if not id or not account_type:
+        return None, None
+    return account_type, id
+
+
+def add_basic_auth_to_url(url: str, username: str, password: str) -> str:
+    if 'https://' in url:
+        return url.replace('https://', f'https://{username}:{password}@')
+    elif 'http:// ' in url:
+        return url.replace('http://', f'http://{username}:{password}@')
+    else:
+        return url

hiddifypanel/models/__init__.py

@@ -1,10 +1,11 @@
-from .config_enum import ConfigCategory, ConfigEnum,Lang
+from .config_enum import ConfigCategory, ConfigEnum, Lang
 from .config import StrConfig, BoolConfig, get_hconfigs, hconfig, set_hconfig, add_or_update_config, bulk_register_configs
 
 from .parent_domain import ParentDomain, add_or_update_parent_domains, bulk_register_parent_domains
 from .domain import Domain, DomainType, ShowDomain, get_domain, get_current_proxy_domains, get_panel_domains, get_proxy_domains, get_proxy_domains_db, get_hdomains, hdomain, add_or_update_domain, bulk_register_domains
 from .proxy import Proxy, ProxyL3, ProxyCDN, ProxyProto, ProxyTransport, add_or_update_proxy, bulk_register_proxies
-from .user import User, UserMode, is_user_active, remaining_days, days_to_reset, user_by_id, user_by_uuid, add_or_update_user,remove_user, user_should_reset, add_or_update_user, bulk_register_users, UserDetail, ONE_GIG
-from .admin import AdminUser, AdminMode, get_super_admin_secret, get_admin_user_db, get_admin_by_uuid, add_or_update_admin, bulk_register_admins, current_admin_or_owner, get_super_admin
+from .user import User, UserMode, is_user_active, remaining_days, days_to_reset, user_by_id, get_user_by_uuid, add_or_update_user, remove_user, user_should_reset, add_or_update_user, bulk_register_users, UserDetail, ONE_GIG, get_user_by_username, get_user_by_username_password
+from .admin import AdminUser, AdminMode, get_super_admin_uuid, get_admin_by_uuid, add_or_update_admin, bulk_register_admins, current_admin_or_owner, get_super_admin, get_admin_by_username, get_admin_by_username_password
 from .child import Child
 from .usage import DailyUsage, get_daily_usage_stats
+from .role import Role