Fix log sanitisation and switch to script.module.bossanova808 logging

.github/workflows/addon-checker.yml

@@ -8,10 +8,10 @@ jobs:
     name: Kodi addon checker
     steps:
     - name: Checkout
-      uses: actions/checkout@v1
+      uses: actions/checkout@v4
     - name: Kodi addon checker validation
       id: kodi-addon-checker
-      uses: xbmc/action-kodi-addon-checker@v1.1
+      uses: xbmc/action-kodi-addon-checker@v1.2
       with:
         kodi-version: matrix
         is-pr: false

resources/language/resource.language.en_gb/strings.po

@@ -27,28 +27,36 @@ msgstr ""
 
 msgctxt "#32024"
 msgid "Android crashlogs are not supported, sorry"
-msgstr "
+msgstr ""
 
 msgctxt "#32025"
 msgid "No log files found to copy ?!"
-msgstr "
+msgstr ""
 
 msgctxt "#32026"
 msgid "Error copying logs"
-msgstr "
+msgstr ""
 
 msgctxt "#32027"
 msgid "No destination path set in the addon settings!"
-msgstr "
+msgstr ""
 
 msgctxt "#32028"
 msgid "Log files copied"
-msgstr "
+msgstr ""
 
 msgctxt "#32029"
 msgid "Something went wrong, (ironically) check your logs!"
-msgstr "
+msgstr ""
 
 msgctxt "#32030"
 msgid "Working..."
-msgstr "
+msgstr ""
+
+msgctxt "#32031"
+msgid "Error making directory to copy log files to!"
+msgstr ""
+
+msgctxt "#32032"
+msgid "Max age of crashlogs to copy (days)"
+msgstr ""

resources/lib/cabertoss.py

@@ -1,31 +1,36 @@
 # -*- coding: utf-8 -*-
 import os
-from datetime import datetime
-from time import sleep
 from datetime import datetime, timedelta
 import socket
+from typing import List, Tuple
 
 import xbmc
 import xbmcvfs
-from bossanova808.constants import *
-from bossanova808.utilities import *
+from bossanova808.constants import LOG_PATH
+from bossanova808.constants import LANGUAGE
 from bossanova808.logger import Logger
 from bossanova808.notify import Notify
 from resources.lib.store import Store
-from resources.lib.clean import *
+from resources.lib.clean import clean_log
 
 
-def gather_log_files():
+def _vfs_join(base: str, name: str) -> str:
+    if base.startswith(('special://', 'smb://', 'nfs://', 'ftp://', 'http://', 'https://')):
+        return base.rstrip('/') + '/' + name
+    return os.path.join(base, name)
+
+
+def gather_log_files() -> List[Tuple[str, str]]:
     """
     Gather a list of the standard Kodi log files (Kodi.log, Kodi.old.log) and the latest crash log, if there is one.
 
-    @return: list of log files in form [type, path], where type is log, oldlog, or crashlog
+    @return: list of log files as (type, path) tuples, where type is 'log', 'oldlog', or 'crashlog'
     """
 
     # Basic log files
-    log_files = [['log', os.path.join(LOG_PATH, 'kodi.log')]]
-    if os.path.exists(os.path.join(LOG_PATH, 'kodi.old.log')):
-        log_files.append(['oldlog', os.path.join(LOG_PATH, 'kodi.old.log')])
+    log_files = [('log', os.path.join(LOG_PATH, 'kodi.log'))]
+    if xbmcvfs.exists(os.path.join(LOG_PATH, 'kodi.old.log')):
+        log_files.append(('oldlog', os.path.join(LOG_PATH, 'kodi.old.log')))
 
     # Can we find a crashlog?
     # @TODO - add Android support if possible..?
@@ -50,7 +55,7 @@ def gather_log_files():
         filematch = 'kodi_'
     elif xbmc.getCondVisibility('system.platform.android'):
         Logger.info("System is Android")
-        Logger.info(LANGUAGE(32023))
+        Logger.info(LANGUAGE(32024))
 
     # If *ELEC, we can be more specific
     if xbmc.getCondVisibility('System.HasAddon(service.coreelec.settings)') or xbmc.getCondVisibility('System.HasAddon(service.libreelec.settings)'):
@@ -59,18 +64,16 @@ def gather_log_files():
         filematch = 'kodi_crashlog_'
 
     if crashlog_path and os.path.isdir(crashlog_path):
-        lastcrash = None
         dirs, possible_crashlog_files = xbmcvfs.listdir(crashlog_path)
         for item in possible_crashlog_files:
             item_with_path = os.path.join(crashlog_path, item)
             if filematch in item and os.path.isfile(item_with_path):
-                if filematch in item:
-                    # Don't bother with older crashlogs
-                    three_days_ago = datetime.now() - timedelta(days=3)
-                    if three_days_ago < datetime.fromtimestamp(os.path.getmtime(item_with_path)):
-                        items.append(os.path.join(crashlog_path, item))
+                # Don't bother with older crashlogs
+                x_days_ago = datetime.now() - timedelta(days=Store.crashlog_max_days)
+                if x_days_ago < datetime.fromtimestamp(os.path.getmtime(item_with_path)):
+                    items.append(os.path.join(crashlog_path, item))
 
-        items.sort(key=lambda f: os.path.getmtime(f))
+        items.sort(key=lambda f:os.path.getmtime(f))
         # Windows crashlogs are a dmp and stacktrace combo...
         if xbmc.getCondVisibility('system.platform.windows'):
             lastcrash = items[-2:]
@@ -80,43 +83,59 @@ def gather_log_files():
         if lastcrash:
             # Logger.info(f"lastcrash {lastcrash}")
             for crashfile in lastcrash:
-                log_files.append(['crashlog', crashfile])
+                log_files.append(('crashlog', crashfile))
 
-    Logger.info("Found these log files to copy:")
-    Logger.info(log_files)
+    Logger.info("Found these log files to copy (type, basename):")
+    Logger.info([[t, os.path.basename(p)] for t, p in log_files])
 
     return log_files
 
 
-def copy_log_files(log_files: []):
+def copy_log_files(log_files: List[Tuple[str, str]]) -> bool:
     """
-    Actually copy the log files to the path in the addon settings
+    Copy the provided Kodi log files into a timestamped destination folder under the configured addon destination.
+
+    Detailed behavior:
+    - Expects log_files as List[Tuple[str, str]] where `type` is e.g. 'log', 'oldlog', or 'crashlog' and `path` is the source filesystem path.
+    - Creates a destination directory at Store.destination_path named "<hostname>_Kodi_Logs_<YYYY-MM-DD_HH-MM-SS>".
+    - For entries with type 'log' or 'oldlog', reads the source, sanitises the content with clean_log() (because the log content may contain URLs with embedded user/password details), and writes the sanitised content to a file with the same basename in the destination folder.
+    - For other types (e.g., crash logs), copies the source file to the destination folder unchanged.
 
-    @param log_files: [] list of log files to copy
-    @return: None
+    Parameters:
+        log_files (List[Tuple[str, str]]): list of log descriptors [type, path] to copy.
+
+    Returns:
+        bool: True if files were successfully copied, False otherwise.
     """
     if not log_files:
         Logger.error(LANGUAGE(32025))
         Notify.error(LANGUAGE(32025))
-        return
+        return False
 
     now_folder_name = f"{socket.gethostname()}_Kodi_Logs_{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}"
-    now_destination_path = os.path.join(Store.destination_path, now_folder_name)
+    now_destination_path = _vfs_join(Store.destination_path, now_folder_name)
 
     try:
         Logger.info(f'Making destination folder: {now_destination_path}')
-        xbmcvfs.mkdir(now_destination_path)
+        if not xbmcvfs.mkdirs(now_destination_path):
+            Logger.error(f'Failed to create destination folder: {now_destination_path}')
+            Notify.error(LANGUAGE(32031))
+            return False
         for file in log_files:
             if file[0] in ['log', 'oldlog']:
                 Logger.info(f'Copying sanitised {file[0]} {file[1]}')
-                with open(xbmcvfs.translatePath(file[1]), 'r', encoding='utf-8') as current:
+                with open(xbmcvfs.translatePath(file[1]), 'r', encoding='utf-8', errors='replace') as current:
                     content = current.read()
                     sanitised = clean_log(content)
-                with xbmcvfs.File(os.path.join(xbmcvfs.translatePath(now_destination_path),os.path.basename(file[1])), 'w') as output:
-                    output.write(sanitised)
+                dest_path = _vfs_join(now_destination_path, os.path.basename(file[1]))
+                f = xbmcvfs.File(dest_path, 'w')
+                try:
+                    f.write(sanitised.encode('utf-8'))
+                finally:
+                    f.close()
             else:
                 Logger.info(f'Copying {file[0]} {file[1]}')
-                if not xbmcvfs.copy(file[1], os.path.join(now_destination_path, os.path.basename(file[1]))):
+                if not xbmcvfs.copy(file[1], _vfs_join(now_destination_path, os.path.basename(file[1]))):
                     return False
         return True
 
@@ -128,18 +147,31 @@ def copy_log_files(log_files: []):
 
 # This is 'main'...
 def run():
-    footprints()
-    Store.load_config_from_settings()
-
-    if not Store.destination_path:
-        Notify.error(LANGUAGE(32027))
-    else:
-        Notify.info(LANGUAGE(32030))
-        log_file_list = gather_log_files()
-        result = copy_log_files(log_file_list)
-        if result:
-            Notify.info(LANGUAGE(32028) + f": {len(log_file_list)}")
+    """
+    Run the log collection and copying flow: initialize this addon's logging, load configuration, gather Kodi log files, copy them to the configured destination, notify the user, and stop this addon's logging.
+
+    This function performs the module's main orchestration. It:
+    - Starts the logger for this addon's internal logging (not Kodi's general logging system) and loads addon configuration from settings.
+    - If no destination path is configured, shows an error notification and skips copying.
+    - Otherwise, notifies the user, gathers available log files, attempts to copy them to the configured destination, and notifies success (including number of files copied) or failure.
+    - Stops this addon's internal logging before returning.
+
+    Side effects: starts/stops this addon's internal logging, reads configuration, performs filesystem operations (reading, sanitizing, and copying log files), and shows user notifications. Returns None.
+    """
+    Logger.start()
+    try:
+        Store.load_config_from_settings()
+
+        if not Store.destination_path:
+            Notify.error(LANGUAGE(32027))
         else:
-            Notify.info(LANGUAGE(32029))
+            Notify.info(LANGUAGE(32030))
+            log_file_list = gather_log_files()
+            result = copy_log_files(log_file_list)
+            if result:
+                Notify.info(LANGUAGE(32028) + f": {len(log_file_list)}")
+            else:
+                Notify.error(LANGUAGE(32029))
     # and, we're done...
-    footprints(startup=False)
+    finally:
+        Logger.stop()

resources/lib/clean.py

@@ -1,15 +1,23 @@
 import re
 
 
-def clean_log(content):
+def clean_log(content: str) -> str:
     """
     Remove username/password details from log file content
 
     @param content:
     @return:
     """
-    replaces = (('//.+?:.+?@', '//USER:PASSWORD@'), ('<user>.+?</user>', '<user>USER</user>'), ('<pass>.+?</pass>', '<pass>PASSWORD</pass>'),)
+    replaces = (
+        # Replace only the credentials part between '//' and '@'
+        (r'(?<=//)([^/@:\s]+):([^/@\s]+)@', r'USER:PASSWORD@'),
+        # Also scrub username only (no password)
+        (r'(?<=//)([^/@:\s]+)@', r'USER@'),
+        # Replace XML username/password; keep it local to the tag
+        (r'(?is)<user>[^<]*</user>', r'<user>USER</user>'),
+        (r'(?is)<pass>[^<]*</pass>', r'<pass>PASSWORD</pass>'),
+    )
 
     for pattern, repl in replaces:
-        sanitised = re.sub(pattern, repl, content)
-        return sanitised
+        content = re.sub(pattern, repl, content)
+    return content

resources/lib/store.py

@@ -1,6 +1,6 @@
-from bossanova808.constants import *
+from bossanova808.constants import ADDON
 from bossanova808.logger import Logger
-from resources.lib.clean import *
+from resources.lib.clean import clean_log
 
 
 class Store:
@@ -13,7 +13,8 @@ class Store:
 
     # Static class variables, referred to elsewhere by Store.whatever
     # https://docs.python.org/3/faq/programming.html#how-do-i-create-static-class-data-and-static-class-methods
-    destination_path = None
+    destination_path: str = ''
+    crashlog_max_days: int = 3
 
     def __init__(self):
         """
@@ -24,15 +25,15 @@ def __init__(self):
     @staticmethod
     def load_config_from_settings():
         """
-        Load in the addon settings, at start or reload them if they have been changed
-        Log each setting as it is loaded
+        Load the addon's configuration from persistent settings.
+
+        Reads the 'log_path' setting and assigns it to Store.destination_path, then logs the resolved path (sanitized with clean_log because these paths may be URLs with embedded user/password details). This is called at startup and when settings are reloaded; it has no return value.
         """
         Logger.info("Loading configuration from settings")
-        Store.destination_path = ADDON.getSetting('log_path')
-
-        Logger.info(f'Logs will be tossed to: {clean_log(Store.destination_path)}')
-
-
-
-
-
+        Store.destination_path = ADDON.getSetting('log_path') or ''
+        if Store.destination_path:
+            Logger.info(f'Logs will be tossed to: {clean_log(Store.destination_path)}')
+        else:
+            Logger.warning('No path set to toss logs to.')
+        Store.crashlog_max_days = int(ADDON.getSetting('crashlog_max_days')) or 3
+        Logger.info(f'Crashlog max days: {Store.crashlog_max_days}')

resources/settings.xml

@@ -17,6 +17,18 @@
                         <heading>32001</heading>
                     </control>
                 </setting>
+                <setting id="crashlog_max_days" type="integer" label="32032" help="">
+                    <level>0</level>
+                    <default>3</default>
+                    <constraints>
+                        <minimum>0</minimum>
+                        <maximum>365</maximum>
+                        <step>1</step>
+                    </constraints>
+                    <control type="edit" format="integer">
+                        <heading>32032</heading>
+                    </control>
+                </setting>
             </group>
         </category>
     </section>