✏️ Editing: myimunify_id.py Read Only

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.error(
                "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, PermissionError) as e:
        logger.error("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)
        text = data.decode("utf-8")
    except UnicodeDecodeError:
        raise MyImunifyIdError
    finally:
        os.close(fd)
    return _parse_id(text)

def _parse_id(text: str) -> str:
    """Read line by line: skip comments (#). First non-comment line must be valid id; nothing after it."""
    id_line = None
    for line in text.splitlines():
        s = line.strip()
        if not s:
            continue
        if s.startswith("#"):
            continue
        if id_line is not None:
            raise MyImunifyIdError
        if len(s) != _ID_LEN or not all(c in _HEX for c in s):
            raise MyImunifyIdError
        id_line = s
    if id_line is None:
        raise MyImunifyIdError
    return id_line

async def _get_myimunify_id_file(user: str) -> Path:
    """Get a file with MyImunify id and create it if does not exist"""

try:
        user_pwd = pwd.getpwnam(user)
    except KeyError as e:
        logger.error("No such user: %s", user)
        raise MyImunifyIdError from e
    else:
        id_file = Path(user_pwd.pw_dir) / MYIMUNIFY_ID_FILE_NAME
        try:
            safe_fileops.ensure_regular_file(str(id_file))
        except FileNotFoundError:
            if not id_file.parent.exists():
                logger.error("No such user homedir: %s", user)
                raise MyImunifyIdError
            try:
                await safe_fileops.touch(str(id_file))
            except (PermissionError, OSError) as e:
                logger.error(
                    "Unable to put myimunify_id in user home dir: %s", e
                )
                raise MyImunifyIdError from e
        except OSError:
            logger.error("Cannot access identity file: %s", id_file)
            raise MyImunifyIdError
    return id_file

🔐 Sid Gifari File Manager Pro