import hashlib import hmac import requests from pypos.core.base_service import BaseService from pypos.core.utils.http_retry import request_with_retry from pypos.core.utils.config_utils import ( CONFIG_FILE, get_http_retry_settings, read_endpoint_config, ) from pypos.modules.auth.config.auth_config import ( get_auth_endpoint_config, ) from pypos.core.utils.ui_message_utils import sanitize_ui_message class AuthService(BaseService): """Service untuk logika bisnis autentikasi""" def __init__(self, http_client=None): super().__init__(http_client=http_client or requests) @staticmethod # edited by glg def hash_password(password): """Hash password menggunakan MD5 (mode permanen).""" return AuthService._hash_password_md5(password) @staticmethod def _hash_password_md5(password): """Hash password menggunakan MD5.""" raw = str(password or "") # edited by glg # Wajib kompatibel dengan hash legacy endpoint/runtime yang masih MD5. return hashlib.md5(raw.encode("utf-8")).hexdigest() # nosec B324 @staticmethod def _is_md5_hash(value): if not isinstance(value, str): return False if len(value) != 32: return False return all(c in "0123456789abcdef" for c in value.lower()) @staticmethod def verify_password(password, password_hash): """Verifikasi password dengan hash MD5.""" ok, _ = AuthService.verify_and_upgrade(password, password_hash) return ok @staticmethod # edited by glg def verify_and_upgrade(password, password_hash): """ Verifikasi password MD5. Return: (bool_ok, new_hash_or_None) """ if not password_hash: return False, None if isinstance(password_hash, bytes): password_hash = password_hash.decode(errors="ignore") if AuthService._is_md5_hash(password_hash): expected = AuthService._hash_password_md5(password) actual = str(password_hash or "").strip().lower() # edited by glg # Gunakan constant-time compare untuk meminimalkan timing leak di jalur verifikasi hash. ok = hmac.compare_digest(expected, actual) return ok, None return False, None @classmethod def is_online(cls): """Cek koneksi internet ke server""" config = get_auth_endpoint_config() base_url = config.get("api_base_url", "") timeout = config.get("request_timeout", 3) if not base_url: return False max_retry, backoff_sec = get_http_retry_settings(config) # edited by glg # Probe online tidak boleh false-negative hanya karena HEAD ditolak server/firewall. # Aturan: # - Jika ada respons HTTP apa pun dari server (termasuk 401/403/404), koneksi dianggap online. # - Jika HEAD tidak didukung (405/501), fallback ke GET. # - Hanya kegagalan koneksi/timeout total yang dianggap offline. try: response = request_with_retry( "HEAD", base_url, timeout=timeout, max_retry=max_retry, backoff_sec=backoff_sec, session=requests, retry_on=(requests.RequestException,), request_kwargs={"allow_redirects": True, "stream": True}, ) return bool(response is not None) except requests.exceptions.HTTPError as exc: status_code = getattr(getattr(exc, "response", None), "status_code", None) if status_code in (405, 501): pass elif status_code is not None: return True else: return False except requests.RequestException: return False try: response = request_with_retry( "GET", base_url, timeout=timeout, max_retry=max_retry, backoff_sec=backoff_sec, session=requests, retry_on=(requests.RequestException,), request_kwargs={"allow_redirects": True, "stream": True}, ) return bool(response is not None) except requests.exceptions.HTTPError as exc: status_code = getattr(getattr(exc, "response", None), "status_code", None) return status_code is not None except requests.RequestException: return False # edited by glg @staticmethod def get_runtime_endpoint_context(): cfg = read_endpoint_config() or {} return { "api_base_url": str(cfg.get("api_base_url") or "").strip(), "config_path": str(CONFIG_FILE or "").strip(), } @staticmethod def classify_login_exception(exc): error = exc status_code = None raw_message = str(error or "").strip() friendly = "" level = "critical" if isinstance(error, requests.exceptions.HTTPError): status_code = getattr(getattr(error, "response", None), "status_code", None) try: status_code = int(status_code) if status_code is not None else None except (TypeError, ValueError): status_code = None if status_code == 401: return { "level": "warning", "status_code": 401, "message": "Username/password salah atau akun belum aktif.", "raw_message": raw_message, } if status_code == 403: return { "level": "warning", "status_code": 403, "message": "Akses login ditolak server. Pastikan akun Anda diizinkan.", "raw_message": raw_message, } if status_code == 404: return { "level": "critical", "status_code": 404, "message": "Endpoint login tidak ditemukan. Periksa Web Admin pada Config.", "raw_message": raw_message, } if status_code in (405, 501): return { "level": "warning", "status_code": status_code, "message": ( "Endpoint login JWT belum tersedia pada server ini " f"(HTTP {status_code})." ), "raw_message": raw_message, } if status_code is not None and status_code >= 500: return { "level": "critical", "status_code": status_code, "message": ( f"Endpoint login bermasalah (HTTP {status_code}). " "Silakan coba lagi beberapa saat." ), "raw_message": raw_message, } if status_code is not None: friendly = f"Login ditolak server (HTTP {status_code})." else: friendly = "Login ditolak server." elif isinstance(error, (requests.exceptions.Timeout, requests.exceptions.ConnectTimeout, requests.exceptions.ReadTimeout)): friendly = "Endpoint login tidak merespons (timeout). Periksa jaringan lalu coba lagi." elif isinstance(error, requests.exceptions.ConnectionError): lower_raw = raw_message.lower() if ( "name or service not known" in lower_raw or "temporary failure in name resolution" in lower_raw or "nodename nor servname provided" in lower_raw or "newconnectionerror" in lower_raw or "max retries exceeded" in lower_raw ): friendly = "Endpoint login tidak dapat dihubungi. Periksa jaringan/DNS lalu coba lagi." else: friendly = "Endpoint login tidak dapat dihubungi. Periksa jaringan lalu coba lagi." elif isinstance(error, requests.exceptions.SSLError): friendly = ( "Koneksi aman ke server gagal (SSL). " "Periksa tanggal/jam perangkat atau sertifikat server." ) else: safe_message, _ = sanitize_ui_message("critical", raw_message) friendly = safe_message or "Terjadi kesalahan saat login." return { "level": level, "status_code": status_code, "message": friendly or "Terjadi kesalahan saat login.", "raw_message": raw_message, }