import threading import requests from PySide6.QtCore import QTimer, Qt from PySide6.QtWidgets import QDialog from pypos.core.base_controller import BaseController from pypos.core.utils.audit_logger import log_audit from pypos.core.utils.config_utils import get_jwt_runtime_config, read_config from pypos.core.utils.device_utils import ( cek_device_ke_server, get_active_device_info, get_device_id, is_device_not_registered, ) from pypos.modules.auth.models.auth_model import AuthModel from pypos.modules.auth.config.auth_config import get_login_network_policy_config from pypos.modules.auth.services.auth_service import AuthService from pypos.modules.auth.services.login_network_policy_service import LoginNetworkPolicyService from pypos.modules.auth.services.login_attempt_guard_service import LoginAttemptGuardService from pypos.modules.auth.services.jwt_auth_service import JwtAuthService from pypos.modules.auth.services.network_probe_service import NetworkProbeService from pypos.modules.auth.services.offline_admin_auth_service import OfflineAdminAuthService from pypos.modules.auth.services.sync_bootstrap_adapter_service import ( SyncBootstrapAdapterService, ) from pypos.modules.auth.views.admin_verification_dialog import AdminVerificationDialog from pypos.core.utils.worker_pool_utils import submit_ui_periodic_task_keyed # edited by glg _NUMERIC_PARSE_EXCEPTIONS = (TypeError, ValueError) _UI_RUNTIME_EXCEPTIONS = (AttributeError, RuntimeError, TypeError, ValueError) _NETWORK_POLICY_EXCEPTIONS = ( RuntimeError, ValueError, TypeError, OSError, ConnectionError, requests.exceptions.RequestException, ) class LoginController(BaseController): def __init__(self, login_view, app): super().__init__() self.log_debug("masuk login controller") self.current_user = None self._last_login_username = "" self._last_login_password = "" self.login_view = login_view self.view = login_view self.app = app self.stacked_widget = app self.model = AuthModel() self.service = AuthService() self.login_attempt_guard_service = LoginAttemptGuardService() self.jwt_auth_service = JwtAuthService() self.offline_admin_auth_service = OfflineAdminAuthService( self.model, self.service, self.login_attempt_guard_service, ) self.sync_bootstrap_adapter_service = SyncBootstrapAdapterService() self.network_probe_service = NetworkProbeService() self.login_network_policy_service = LoginNetworkPolicyService( probe_service=self.network_probe_service ) self._last_network_gate = {} self._network_block_notice_shown = False self._network_policy_timer = None self._network_probe_lock = threading.Lock() self._network_probe_inflight = False if self.login_view and hasattr(self.login_view, "network_gate_payload_ready"): self.login_view.network_gate_payload_ready.connect( self._on_network_gate_payload_ready, Qt.QueuedConnection, ) self._init_network_policy_monitor() # edited by glg @staticmethod def _as_int(value, fallback=3000, minimum=500): try: parsed = int(value) except _NUMERIC_PARSE_EXCEPTIONS: parsed = int(fallback) return max(int(minimum), parsed) # edited by glg def _resolve_notice_level(self, gate): state = str((gate or {}).get("state") or "").strip().upper() blocked = bool((gate or {}).get("block_login")) if state == "PROBING": return "info" if blocked: return "error" if state in { LoginNetworkPolicyService.STATE_ONLINE_SERVER_UNSTABLE, LoginNetworkPolicyService.STATE_ONLINE_INTERNET_ONLY, }: return "warning" return "success" # edited by glg def _build_login_lock_message(self, guard_state): return self.login_attempt_guard_service.build_lock_message("login", guard_state) # edited by glg def _check_login_guard_locked(self, username): safe_username = str(username or "") per_user = self.login_attempt_guard_service.check_attempt( LoginAttemptGuardService.SCOPE_LOGIN, safe_username, ) if per_user.get("locked"): return True, per_user global_state = self.login_attempt_guard_service.check_attempt( LoginAttemptGuardService.SCOPE_LOGIN_GLOBAL, "__global__", ) if global_state.get("locked"): return True, global_state return False, {} # edited by glg def _record_login_failure(self, username): safe_username = str(username or "") self.login_attempt_guard_service.record_failure( LoginAttemptGuardService.SCOPE_LOGIN, safe_username, ) self.login_attempt_guard_service.record_failure( LoginAttemptGuardService.SCOPE_LOGIN_GLOBAL, "__global__", ) return self._check_login_guard_locked(safe_username) # edited by glg def _apply_network_gate_to_view(self, gate, source="timer"): if not isinstance(gate, dict): return message = str(gate.get("message") or "").strip() state = str(gate.get("state") or "").strip() is_probing = state.upper() == "PROBING" blocked = bool(gate.get("block_login")) policy = gate.get("policy") if isinstance(gate.get("policy"), dict) else {} disable_when_blocked = bool(policy.get("disable_button_when_blocked")) allow_login_button = not (blocked and disable_when_blocked) notice_level = self._resolve_notice_level(gate) if self.login_view and hasattr(self.login_view, "update_network_gate"): try: self.login_view.update_network_gate( message=message, level=notice_level, state=state, blocked=blocked, ) except _UI_RUNTIME_EXCEPTIONS as exc: self.log_warning(f"Gagal update status jaringan di login view: {exc}") if self.login_view and hasattr(self.login_view, "set_login_enabled"): try: self.login_view.set_login_enabled(allow_login_button) except _UI_RUNTIME_EXCEPTIONS as exc: self.log_warning(f"Gagal atur enable button login: {exc}") should_show_notice = bool(policy.get("show_startup_notice")) if blocked and should_show_notice and not is_probing and not self._network_block_notice_shown: self._network_block_notice_shown = True self.show_warning( "Koneksi Diperlukan", "Aplikasi harus online untuk login.\nSilakan periksa jaringan lalu coba lagi.", view=self.login_view, ) self.log_warning( f"Login diblokir oleh kebijakan jaringan (source={source}): reason={gate.get('reason')}" ) if not blocked: self._network_block_notice_shown = False # edited by glg def _get_recheck_interval_ms(self, gate=None): policy = gate.get("policy") if isinstance(gate, dict) else {} return self._as_int( policy.get("recheck_interval_ms"), fallback=3000, minimum=500, ) # edited by glg def _set_network_probe_inflight(self, value): with self._network_probe_lock: self._network_probe_inflight = bool(value) # edited by glg def _try_mark_network_probe_inflight(self): with self._network_probe_lock: if self._network_probe_inflight: return False self._network_probe_inflight = True return True # edited by glg def _build_pending_network_gate(self): policy = get_login_network_policy_config() or {} normalized = self.login_network_policy_service._normalize_policy(policy) return { "state": "PROBING", "raw_state": "", "reason": "probing_in_progress", "message": "Memeriksa koneksi jaringan...", "allow_login": False, "block_login": True, "server_online": False, "internet_online": False, "within_grace": False, "fail_streak": 0, "success_streak": 0, "policy": dict(normalized or {}), "probe": {}, "timestamp": 0.0, } # edited by glg def _dispatch_network_policy_refresh_async(self, source="manual", force_probe=False): _ = bool(force_probe) if not self._try_mark_network_probe_inflight(): return False started_event = threading.Event() def _worker(): started_event.set() payload = { "source": str(source or "manual"), "gate": {}, "error": "", } try: payload["gate"] = self.login_network_policy_service.evaluate() except _NETWORK_POLICY_EXCEPTIONS as exc: payload["error"] = str(exc) payload["gate"] = dict(self._last_network_gate or {}) if self.login_view and hasattr(self.login_view, "emit_network_gate_payload"): self.login_view.emit_network_gate_payload(payload) return self._on_network_gate_payload_ready(payload) key = f"login-network-policy:{id(self)}" future = submit_ui_periodic_task_keyed(key, _worker) if future is None: self._set_network_probe_inflight(False) return False # edited by glg # Fail-safe: saat antrean worker penuh, task bisa ter-drop. # Pastikan flag inflight tidak terkunci agar probe berikutnya tetap bisa jalan. def _on_done(_): if not started_event.is_set(): self._set_network_probe_inflight(False) try: future.add_done_callback(_on_done) except _UI_RUNTIME_EXCEPTIONS: if not started_event.is_set(): self._set_network_probe_inflight(False) return True # edited by glg def _on_network_gate_payload_ready(self, payload): self._set_network_probe_inflight(False) data = payload if isinstance(payload, dict) else {} source = str(data.get("source") or "worker").strip() gate = data.get("gate") if isinstance(data.get("gate"), dict) else {} error = str(data.get("error") or "").strip() if error: self.log_warning(f"Gagal evaluasi policy jaringan login (source={source}): {error}") if gate: self._last_network_gate = dict(gate) self._apply_network_gate_to_view(gate, source=source) if self._network_policy_timer is not None: try: next_interval = self._get_recheck_interval_ms(gate=gate) if int(self._network_policy_timer.interval()) != int(next_interval): self._network_policy_timer.setInterval(int(next_interval)) except _UI_RUNTIME_EXCEPTIONS: pass # edited by glg def refresh_network_policy_now(self, source="manual", force_probe=False): gate = self._last_network_gate if isinstance(self._last_network_gate, dict) else {} if not gate: gate = self._build_pending_network_gate() self._apply_network_gate_to_view(gate, source=source) if bool(force_probe) or not bool(self._last_network_gate): self._dispatch_network_policy_refresh_async(source=source, force_probe=force_probe) return gate # edited by glg def get_network_gate(self, force_probe=False, source="runtime"): return self.refresh_network_policy_now( source=source, force_probe=bool(force_probe), ) # edited by glg def _init_network_policy_monitor(self): if not self.login_view: return interval_ms = self._get_recheck_interval_ms() self._network_policy_timer = QTimer(self.login_view) self._network_policy_timer.setInterval(interval_ms) self._network_policy_timer.timeout.connect( lambda: self.refresh_network_policy_now(source="timer", force_probe=True) ) self._network_policy_timer.start() QTimer.singleShot( 0, lambda: self.refresh_network_policy_now(source="startup", force_probe=True), ) # edited by glg def _validate_runtime_endpoint_for_login(self): context = self.service.get_runtime_endpoint_context() base_url = str(context.get("api_base_url") or "").strip() if not base_url: return ( False, "Web Admin belum diatur untuk runtime aplikasi.\n" "Silakan klik Config dan isi Web Admin yang benar.\n" "Sumber runtime: AppData/Roaming/PyPOS/config.json", ) machine_id = get_device_id() local_device = get_active_device_info(machine_id) or {} local_cabang_id = int(local_device.get("cabang_id") or 0) if local_cabang_id <= 0: self.log_warning("Validasi endpoint runtime dilewati karena cabang lokal belum tersedia.") return True, "" try: result = cek_device_ke_server(machine_id) except _NETWORK_POLICY_EXCEPTIONS as exc: self.log_warning(f"Validasi endpoint runtime skip (cek server gagal): {exc}") return True, "" if not isinstance(result, dict): return True, "" raw_status = result.get("status") try: status = int(raw_status) except _NUMERIC_PARSE_EXCEPTIONS: status = raw_status if status == 200: data_reg = result.get("dataReg") if not isinstance(data_reg, dict): data_reg = {} server_cabang_id = int(data_reg.get("cabang_id") or 0) if server_cabang_id > 0 and server_cabang_id != local_cabang_id: return ( False, "Endpoint runtime tidak sesuai dengan device ini.\n" f"Web Admin aktif: {base_url}\n" f"Cabang lokal: {local_cabang_id}, cabang server: {server_cabang_id}\n" "Silakan sesuaikan Web Admin di menu Config.", ) cfg = read_config() or {} local_toko_id = int(cfg.get("toko_id") or 0) server_toko_id = int(data_reg.get("toko_id") or 0) if local_toko_id > 0 and server_toko_id > 0 and local_toko_id != server_toko_id: return ( False, "Endpoint runtime tidak sesuai dengan toko device ini.\n" f"Web Admin aktif: {base_url}\n" f"Toko lokal: {local_toko_id}, toko server: {server_toko_id}\n" "Silakan sesuaikan Web Admin di menu Config.", ) return True, "" if status == 202: return ( False, "Device masih menunggu approval pada endpoint runtime aktif.\n" f"Web Admin aktif: {base_url}", ) if status == 404 or is_device_not_registered(result): return ( False, "Device belum terdaftar pada endpoint runtime aktif.\n" f"Web Admin aktif: {base_url}\n" "Silakan lakukan registrasi device pada endpoint yang benar.", ) return True, "" def proses_login(self, username, password): if not username or not password: self.show_warning("Login Gagal", "Username dan password harus diisi.", view=self.login_view) return self._last_login_username = str(username or "") self._last_login_password = str(password or "") login_locked, lock_state = self._check_login_guard_locked(self._last_login_username) if login_locked: self.show_warning( "Login Ditunda", self._build_login_lock_message(lock_state), view=self.login_view, ) self._last_login_password = "" return jwt_cfg = get_jwt_runtime_config() gate = self.get_network_gate(force_probe=True, source="submit") if bool(gate.get("block_login")): self.show_warning( "Koneksi Diperlukan", str(gate.get("message") or "Aplikasi harus online untuk login."), view=self.login_view, ) self._last_login_password = "" return server_online = bool(gate.get("server_online")) if bool(jwt_cfg.get("enabled")) and bool(jwt_cfg.get("required")) and not server_online: self.show_warning( "Login Gagal", "Server login JWT belum bisa dijangkau. Silakan cek koneksi ke server POS.", view=self.login_view, ) self._last_login_password = "" return if server_online: runtime_ok, runtime_message = self._validate_runtime_endpoint_for_login() if not runtime_ok: self.show_warning("Endpoint Runtime Tidak Sesuai", runtime_message, view=self.login_view) self._last_login_password = "" return if server_online and jwt_cfg.get("enabled"): try: user_info = self._proses_login_online_jwt(username=username, password=password) self.log_info(f"login sukses {user_info}") self.login_attempt_guard_service.reset_attempt( LoginAttemptGuardService.SCOPE_LOGIN, self._last_login_username, ) self.login_success(user_info, token_ready=True) self.login_view.clear_fields() return except (RuntimeError, ValueError, TypeError, OSError, ConnectionError, requests.exceptions.RequestException) as exc: mapped = self.service.classify_login_exception(exc) self.log_warning( f"Login JWT gagal: status={mapped.get('status_code')} raw={mapped.get('raw_message')}" ) if int(mapped.get("status_code") or 0) == 401: failed_locked, failed_state = self._record_login_failure(self._last_login_username) if failed_locked: self.show_warning( "Login Ditunda", self._build_login_lock_message(failed_state), view=self.login_view, ) self._last_login_password = "" return if jwt_cfg.get("required"): status_code = mapped.get("status_code") if status_code in (404, 405, 501): runtime_ctx = self.service.get_runtime_endpoint_context() runtime_url = str(runtime_ctx.get("api_base_url") or "").strip() or "-" runtime_cfg = str(runtime_ctx.get("config_path") or "").strip() or "-" self.show_warning( "Endpoint JWT Tidak Tersedia", ( f"{mapped.get('message')}\n" f"Web Admin runtime: {runtime_url}\n" f"Sumber config runtime: {runtime_cfg}\n\n" "Gunakan endpoint yang sudah mendukung JWT " "atau nonaktifkan `jwt_required` jika endpoint ini memang non-JWT." ), view=self.login_view, ) self._last_login_password = "" return level = str(mapped.get("level") or "critical").lower() message = str(mapped.get("message") or "Terjadi kesalahan saat login.") if level == "warning": self.show_warning("Login Gagal", message, view=self.login_view) else: self.show_error("Login Gagal", message, view=self.login_view) self._last_login_password = "" return self.log_warning(f"Login JWT gagal, fallback lokal: {exc}") user = self.model.get_user_by_username(username) try: ok, new_hash = self.service.verify_and_upgrade(password, user["password"]) if user else (False, None) except RuntimeError as e: self.show_warning("Login Gagal", str(e), view=self.login_view) failed_locked, failed_state = self._record_login_failure(self._last_login_username) if failed_locked: self.show_warning( "Login Ditunda", self._build_login_lock_message(failed_state), view=self.login_view, ) self._last_login_password = "" return if user and ok: if new_hash: self.model.update_password_hash_by_id(user["id"], new_hash) self.login_attempt_guard_service.reset_attempt( LoginAttemptGuardService.SCOPE_LOGIN, self._last_login_username, ) self.login_attempt_guard_service.reset_attempt( LoginAttemptGuardService.SCOPE_LOGIN_GLOBAL, "__global__", ) user_info = {"id": user["id"], "nama": user["nama"]} self.log_info(f"login sukses {user_info}") self.login_success(user_info) self.login_view.clear_fields() else: failed_locked, failed_state = self._record_login_failure(self._last_login_username) if failed_locked: self.show_warning( "Login Ditunda", self._build_login_lock_message(failed_state), view=self.login_view, ) self._last_login_password = "" return self.show_warning("Login Gagal", "Username atau password salah.", view=self.login_view) self._last_login_password = "" def _proses_login_online_jwt(self, username, password): machine_id = get_device_id() cabang_id = "" try: active_device = get_active_device_info(machine_id) or {} cabang_id = active_device.get("cabang_id") or "" except _NETWORK_POLICY_EXCEPTIONS: cabang_id = "" token_bundle = self.jwt_auth_service.issue_token( username=str(username or ""), password=str(password or ""), machine_id=machine_id, ) self.app.session.set_machine_context(machine_id=machine_id, cabang_id=cabang_id) self.app.session.set_jwt_tokens( access_token=token_bundle.get("access_token"), refresh_token=token_bundle.get("refresh_token"), token_type=token_bundle.get("token_type"), expires_in=token_bundle.get("expires_in"), ) # edited by glg # Sinkronisasi per_employee saat login dinonaktifkan. # Seed akun dilakukan saat device sudah approved (sebelum login pertama). user = self.model.get_user_by_username(username) if not user: raise RuntimeError( "Akun belum tersedia di perangkat ini. " "Pastikan device sudah approved dan sinkronisasi akun awal berhasil." ) return {"id": user["id"], "nama": user["nama"]} def _sync_per_employee_after_jwt(self, machine_id, cabang_id): self.sync_bootstrap_adapter_service.sync_per_employee( machine_id=machine_id, cabang_id=cabang_id, ) def login_success(self, user_info, token_ready=False): self.log_debug("masuk login controller.py") self.current_user = user_info gate = self.get_network_gate(force_probe=False, source="login_success") policy = gate.get("policy") if isinstance(gate.get("policy"), dict) else {} if bool(gate.get("block_login")): allow_offline_emergency = ( str(gate.get("state") or "").strip().upper() == LoginNetworkPolicyService.STATE_OFFLINE and bool(policy.get("offline_emergency_admin_enabled")) ) if allow_offline_emergency: self.log_debug("cek mode offline darurat") jawab = self.ask_confirm( "Mode Offline", "Koneksi internet tidak tersedia.\n" "POS ini membutuhkan koneksi internet saat login.\n\n" "Apakah Anda ingin login dalam mode darurat offline dengan persetujuan Admin?", view=self.login_view, ) if jawab: if self.otentikasi_admin(): self.show_warning( "PERHATIAN!", "Anda sedang login dalam kondisi offline.\n" "Beberapa fitur mungkin tidak tersedia:\n" "- Data produk tidak update\n" "- Harga atau diskon bisa saja tidak akurat\n" "- Data penjualan tidak akan terkirim ke server\n\n" "Gunakan data terakhir yang tersedia di sistem lokal.", view=self.login_view, ) self.current_user["mode"] = "offline" log_audit( self.model.db_path, self.current_user["id"], "INSERT", "login offline", self.current_user["id"], "Login offline dengan admin berhasil", ) self.lanjutkan_login_offline() self._last_login_password = "" return self.show_error("Gagal", "Otorisasi admin gagal.", view=self.login_view) self._last_login_password = "" return self._last_login_password = "" return self.show_warning( "Koneksi Diperlukan", str(gate.get("message") or "Login ditahan karena koneksi belum memenuhi kebijakan."), view=self.login_view, ) self._last_login_password = "" return jwt_cfg = get_jwt_runtime_config() if jwt_cfg.get("enabled") and not token_ready: if not bool(gate.get("server_online")): if jwt_cfg.get("required"): self.show_error( "Login Gagal", "Token JWT wajib, tetapi server tidak dapat dijangkau saat ini.", view=self.login_view, ) self._last_login_password = "" return else: machine_id = get_device_id() cabang_id = "" try: active_device = get_active_device_info(machine_id) or {} cabang_id = active_device.get("cabang_id") or "" except _NETWORK_POLICY_EXCEPTIONS: cabang_id = "" try: token_bundle = self.jwt_auth_service.issue_token( username=self._last_login_username, password=self._last_login_password, machine_id=machine_id, ) self.app.session.set_machine_context(machine_id=machine_id, cabang_id=cabang_id) self.app.session.set_jwt_tokens( access_token=token_bundle.get("access_token"), refresh_token=token_bundle.get("refresh_token"), token_type=token_bundle.get("token_type"), expires_in=token_bundle.get("expires_in"), ) except (RuntimeError, ValueError, TypeError, OSError, ConnectionError, requests.exceptions.RequestException) as exc: if jwt_cfg.get("required"): self.show_error( "Login Gagal", f"Gagal mendapatkan token JWT: {str(exc)}", view=self.login_view, ) self._last_login_password = "" return self.log_warning(f"JWT tidak aktif pada sesi ini: {exc}") self.app.login_success(user_info) self._last_login_password = "" def otentikasi_admin(self): while True: dialog = AdminVerificationDialog(self.login_view) if dialog.exec() != QDialog.Accepted: return False username, password = dialog.get_credentials() ok, message, _ = self.offline_admin_auth_service.validate_admin_credentials(username, password) if ok: return True self.show_warning("Gagal", message, view=self.login_view) def cek_admin_lokal(self, username, password): try: ok, _, admin = self.offline_admin_auth_service.validate_admin_credentials(username, password) if ok and admin: user_info = {"id": admin["id"], "nama": admin["nama"]} self.log_info(f"LoginController: login offline sukses {user_info}") return True except (RuntimeError, ValueError, TypeError, OSError) as e: self.log_error(f"Error cek admin: {e}") return False def lanjutkan_login_offline(self): self.app.login_success(self.current_user) def show_dashboard(self, user_info): self.app.show_dashboard(user_info)