C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// import os import pwd import stat import uuid from pathlib import Path from typing import Dict, List, Optional from defence360agent.contracts.permissions import logger from defence360agent.model import instance from defence360agent.myimunify.model import MyImunify, update_users_protection from defence360agent.subsys.panels.hosting_panel import HostingPanel from defence360agent.utils import safe_fileops MYIMUNIFY_ID_FILE_NAME = ".myimunify_id" _BANNER = ( "# DO NOT EDIT\n# This file contains MyImunify id unique to this user\n\n" ) _ID_LEN = 32 _HEX = frozenset("0123456789abcdef") class MyImunifyIdError(Exception): """Exception representing issues related to MyImunify id""" async def add_myimunify_user( sink, user: str, protection: bool ) -> Optional[str]: """Save subscription type to the DB and generate id file""" myimunify, _ = MyImunify.get_or_create(user=user) myimunify.save() await update_users_protection(sink, [user], protection) logger.info("Applied setting MyImunify=%s for user %s", protection, user) try: myimunify_id = await _get_or_generate_id(user) except MyImunifyIdError: # User no longer exists return None return myimunify_id async def get_myimunify_users() -> List[Dict]: """ Get a list of MyImunify users, their subscription types and unique ids """ users = [] user_details = await HostingPanel().get_user_details() myimunify_user_to_id = await _myimunify_user_to_id() with instance.db.transaction(): for user, myimunify_uid in sorted(myimunify_user_to_id.items()): record, _ = MyImunify.get_or_create(user=user) users.append( { "email": user_details.get(user, {}).get("email", ""), "username": user, "myimunify_id": myimunify_uid, "protection": record.protection, "locale": user_details.get(user, {}).get("locale", ""), } ) return users async def _myimunify_user_to_id() -> Dict[str, str]: """Get a list of users and their MyImunify ids""" user_to_id = {} for user in await HostingPanel().get_users(): try: user_to_id[user] = await _get_or_generate_id(user) except MyImunifyIdError: # User does not exist continue except safe_fileops.UnsafeFileOperation as e: logger.warning( "Unable to generate id for user=%s, error=%s", user, str(e) ) continue return user_to_id async def _get_or_generate_id(user: str) -> str: """ Read MyImunify id if exists and valid, or generate a new one and write into the file. Malformed files are regenerated. """ id_file = await _get_myimunify_id_file(user) try: return _read_id(id_file) except (FileNotFoundError, MyImunifyIdError): myimunify_id = uuid.uuid1().hex return await _write_id(myimunify_id, id_file) async def _write_id(myimunify_id: str, id_file: Path) -> str: """Write MyImunify id to file""" text = _BANNER + myimunify_id + "\n" try: await safe_fileops.write_text(str(id_file), text) except OSError as e: logger.warning("Unable to write myimunify_id in user home dir: %s", e) raise MyImunifyIdError from e return myimunify_id def _read_id(id_file: Path) -> str: """Read and validate MyImunify id from file. Raises MyImunifyIdError if malformed. Opens with O_RDONLY | O_NONBLOCK and verifies via fstat() that the fd refers to a regular file before reading. This eliminates the TOCTOU window between a path-level type check and the actual read (e.g. an attacker replacing the file with a FIFO between the two). """ try: fd = os.open(str(id_file), os.O_RDONLY | os.O_NONBLOCK) except FileNotFoundError: raise except OSError: raise MyImunifyIdError try: if not stat.S_ISREG(os.fstat(fd).st_mode): raise MyImunifyIdError data = os.read(fd, 8192) t