404

[ Avaa Bypassed ]




Upload:

Command:

botdev@18.225.156.49: ~ $
import json
import logging
import os
from collections import OrderedDict
from typing import Any, Dict, List  # noqa: F401

from uaclient import defaults, system, util


class RedactionFilter(logging.Filter):
    """A logging filter to redact confidential info"""

    def filter(self, record: logging.LogRecord):
        record.msg = util.redact_sensitive_logs(str(record.msg))
        return True


class JsonArrayFormatter(logging.Formatter):
    """Json Array Formatter for our logging mechanism
    Custom made for Pro logging needs
    """

    default_time_format = "%Y-%m-%dT%H:%M:%S"
    default_msec_format = "%s.%03d"
    required_fields = (
        "asctime",
        "levelname",
        "name",
        "funcName",
        "lineno",
        "message",
    )

    def format(self, record: logging.LogRecord) -> str:
        record.message = record.getMessage()
        record.asctime = self.formatTime(record)

        extra_message_dict = {}  # type: Dict[str, Any]
        if record.exc_info:
            extra_message_dict["exc_info"] = self.formatException(
                record.exc_info
            )
        if not extra_message_dict.get("exc_info") and record.exc_text:
            extra_message_dict["exc_info"] = record.exc_text
        if record.stack_info:
            extra_message_dict["stack_info"] = self.formatStack(
                record.stack_info
            )
        extra = record.__dict__.get("extra")
        if extra and isinstance(extra, dict):
            extra_message_dict.update(extra)

        # is ordered to maintain order of fields in log output
        local_log_record = OrderedDict()  # type: Dict[str, Any]
        # update the required fields in the order stated
        for field in self.required_fields:
            value = record.__dict__.get(field)
            local_log_record[field] = value

        local_log_record["extra"] = extra_message_dict
        return json.dumps(list(local_log_record.values()))


def get_user_log_file() -> str:
    """Gets the correct user log_file storage location"""
    return system.get_user_cache_dir() + "/ubuntu-pro.log"


def get_all_user_log_files() -> List[str]:
    """Gets all the log files for the users in the system

    Returns a list of all user log files in their home directories.
    """
    user_directories = os.listdir("/home")
    log_files = []
    for user_directory in user_directories:
        user_path = (
            "/home/"
            + user_directory
            + "/.cache/"
            + defaults.USER_CACHE_SUBDIR
            + "/ubuntu-pro.log"
        )
        if os.path.isfile(user_path):
            log_files.append(user_path)
    return log_files


def setup_journald_logging(log_level, logger):
    logger.setLevel(log_level)
    logger.addFilter(RedactionFilter())
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(JsonArrayFormatter())
    console_handler.setLevel(log_level)
    logger.addHandler(console_handler)

Filemanager

Name Type Size Permission Actions
__pycache__ Folder 0755
api Folder 0755
cli Folder 0755
clouds Folder 0755
daemon Folder 0755
entitlements Folder 0755
files Folder 0755
http Folder 0755
messages Folder 0755
timer Folder 0755
__init__.py File 0 B 0644
actions.py File 8.79 KB 0644
apt.py File 32.73 KB 0644
apt_news.py File 6.54 KB 0644
config.py File 23.1 KB 0644
contract.py File 30.3 KB 0644
contract_data_types.py File 9.89 KB 0644
data_types.py File 10.27 KB 0644
defaults.py File 2.1 KB 0644
event_logger.py File 8.06 KB 0644
exceptions.py File 14.62 KB 0644
gpg.py File 836 B 0644
livepatch.py File 12.51 KB 0644
lock.py File 3.56 KB 0644
log.py File 2.91 KB 0644
security.py File 56.78 KB 0644
security_status.py File 25.25 KB 0644
snap.py File 6.84 KB 0644
status.py File 29.69 KB 0644
system.py File 24.65 KB 0644
types.py File 308 B 0644
upgrade_lts_contract.py File 3.5 KB 0644
util.py File 15.31 KB 0644
version.py File 2.63 KB 0644
yaml.py File 840 B 0644