{Security} Set log file permissions to 600 on creation

src/azure-cli-core/azure/cli/core/azlogging.py

@@ -26,6 +26,7 @@
 import os
 import logging
 import datetime
+from logging.handlers import RotatingFileHandler
 
 from azure.cli.core.commands.events import EVENT_INVOKER_PRE_CMD_TBL_TRUNCATE
 
@@ -38,6 +39,20 @@
 _CMD_LOG_LINE_PREFIX = "CMD-LOG-LINE-BEGIN"
 
 
+class SecureFileHandler(logging.FileHandler):
+    """A FileHandler that creates the log file with 600 permissions (owner read/write only)."""
+    def _open(self):
+        fd = os.open(self.baseFilename, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
+        return os.fdopen(fd, self.mode, encoding=self.encoding)
+
+
+class SecureRotatingFileHandler(RotatingFileHandler):
+    """A RotatingFileHandler that creates log files with 600 permissions (owner read/write only)."""
+    def _open(self):
+        fd = os.open(self.baseFilename, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
+        return os.fdopen(fd, self.mode, encoding=self.encoding)
+
+
 class AzCliLogging(CLILogging):
     COMMAND_METADATA_LOGGER = 'az_command_data_logger'
 
@@ -58,6 +73,20 @@ def configure(self, args):
             # when debug log is shown.
             logging.getLogger("azure.core.pipeline.policies.http_logging_policy").setLevel(logging.CRITICAL)
 
+    def _init_logfile_handlers(self, root_logger, cli_loggers):
+        # Override knack's CLILogging._init_logfile_handlers() (called by CLILogging.configure())
+        # to use SecureRotatingFileHandler, ensuring log files are created with 0o600 permissions.
+        ensure_dir(self.log_dir)
+        log_file_path = os.path.join(self.log_dir, self.logfile_name)
+        logfile_handler = SecureRotatingFileHandler(log_file_path, maxBytes=10 * 1024 * 1024, backupCount=5,
+                                                    encoding=LOG_FILE_ENCODING)
+        lfmt = logging.Formatter('%(process)d : %(asctime)s : %(levelname)s : %(name)s : %(message)s')
+        logfile_handler.setFormatter(lfmt)
+        logfile_handler.setLevel(logging.DEBUG)
+        root_logger.addHandler(logfile_handler)
+        for cli_logger in cli_loggers:
+            cli_logger.addHandler(logfile_handler)
+
     def get_command_log_dir(self):
         return self.command_log_dir
 
@@ -112,7 +141,7 @@ def _init_command_logfile_handlers(self, command_metadata_logger, args):
         log_file_path = os.path.join(self.command_log_dir, log_name)
         get_logger(__name__).debug("metadata file logging enabled - writing logs to '%s'.", log_file_path)
 
-        logfile_handler = logging.FileHandler(log_file_path, encoding=LOG_FILE_ENCODING)
+        logfile_handler = SecureFileHandler(log_file_path, encoding=LOG_FILE_ENCODING)
 
         lfmt = logging.Formatter(_CMD_LOG_LINE_PREFIX + ' %(process)d | %(asctime)s | %(levelname)s | %(name)s | %(message)s')  # pylint: disable=line-too-long
         logfile_handler.setFormatter(lfmt)

src/azure-cli-core/azure/cli/core/tests/test_azlogging.py

@@ -0,0 +1,132 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+import sys
+import stat
+import logging
+import tempfile
+import unittest
+
+from azure.cli.core.azlogging import SecureFileHandler, SecureRotatingFileHandler
+
+
+class TestSecureFileHandler(unittest.TestCase):
+
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        import shutil
+        shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+    def test_secure_file_handler_writes_log(self):
+        log_path = os.path.join(self.temp_dir, 'test.log')
+        handler = SecureFileHandler(log_path)
+        handler.setFormatter(logging.Formatter('%(message)s'))
+        logger = logging.getLogger('test_secure_file_handler')
+        logger.addHandler(handler)
+        logger.setLevel(logging.DEBUG)
+        try:
+            logger.info('test message')
+            handler.flush()
+            with open(log_path, 'r') as f:
+                self.assertIn('test message', f.read())
+        finally:
+            logger.removeHandler(handler)
+            handler.close()
+
+    @unittest.skipIf(sys.platform == 'win32', 'POSIX file permissions not applicable on Windows')
+    def test_secure_file_handler_permissions(self):
+        log_path = os.path.join(self.temp_dir, 'secure.log')
+        handler = SecureFileHandler(log_path)
+        handler.setFormatter(logging.Formatter('%(message)s'))
+        logger = logging.getLogger('test_secure_file_handler_perms')
+        logger.addHandler(handler)
+        logger.setLevel(logging.DEBUG)
+        try:
+            logger.info('secret data')
+            handler.flush()
+            mode = os.stat(log_path).st_mode
+            self.assertEqual(stat.S_IMODE(mode), 0o600)
+        finally:
+            logger.removeHandler(handler)
+            handler.close()
+
+
+class TestSecureRotatingFileHandler(unittest.TestCase):
+
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        import shutil
+        shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+    def test_secure_rotating_file_handler_writes_log(self):
+        log_path = os.path.join(self.temp_dir, 'rotating.log')
+        handler = SecureRotatingFileHandler(log_path, maxBytes=1024, backupCount=2)
+        handler.setFormatter(logging.Formatter('%(message)s'))
+        logger = logging.getLogger('test_secure_rotating_handler')
+        logger.addHandler(handler)
+        logger.setLevel(logging.DEBUG)
+        try:
+            logger.info('rotating test message')
+            handler.flush()
+            with open(log_path, 'r') as f:
+                self.assertIn('rotating test message', f.read())
+        finally:
+            logger.removeHandler(handler)
+            handler.close()
+
+    @unittest.skipIf(sys.platform == 'win32', 'POSIX file permissions not applicable on Windows')
+    def test_secure_rotating_file_handler_permissions(self):
+        log_path = os.path.join(self.temp_dir, 'rotating_secure.log')
+        handler = SecureRotatingFileHandler(log_path, maxBytes=1024, backupCount=2)
+        handler.setFormatter(logging.Formatter('%(message)s'))
+        logger = logging.getLogger('test_secure_rotating_handler_perms')
+        logger.addHandler(handler)
+        logger.setLevel(logging.DEBUG)
+        try:
+            logger.info('secret rotating data')
+            handler.flush()
+            mode = os.stat(log_path).st_mode
+            self.assertEqual(stat.S_IMODE(mode), 0o600)
+        finally:
+            logger.removeHandler(handler)
+            handler.close()
+
+    @unittest.skipIf(sys.platform == 'win32', 'POSIX file permissions not applicable on Windows')
+    def test_secure_rotating_file_handler_rotation_preserves_permissions(self):
+        log_path = os.path.join(self.temp_dir, 'rotate_perm.log')
+        handler = SecureRotatingFileHandler(log_path, maxBytes=50, backupCount=2)
+        handler.setFormatter(logging.Formatter('%(message)s'))
+        logger = logging.getLogger('test_secure_rotating_rotation')
+        logger.addHandler(handler)
+        logger.setLevel(logging.DEBUG)
+        try:
+            # Write enough data to trigger at least one rotation
+            for i in range(20):
+                logger.info('message %d with enough length to rotate', i)
+            handler.flush()
+
+            # Check that the main log file has 0o600 permissions
+            mode = os.stat(log_path).st_mode
+            self.assertEqual(stat.S_IMODE(mode), 0o600)
+
+            # Check backup files have 0o600 permissions too
+            for suffix in ['.1', '.2']:
+                backup = log_path + suffix
+                if os.path.exists(backup):
+                    mode = os.stat(backup).st_mode
+                    self.assertEqual(stat.S_IMODE(mode), 0o600,
+                                     f"Backup file {backup} has wrong permissions")
+        finally:
+            logger.removeHandler(handler)
+            handler.close()
+
+
+if __name__ == '__main__':
+    unittest.main()

src/azure-cli/azure/cli/command_modules/appservice/tests/latest/test_webapp_commands.py

@@ -161,7 +161,7 @@ def test_win_webapp_quick_create_runtime(self, resource_group):
         webapp_name_2 = self.create_random_name(prefix='webapp-quick', length=24)
         plan = self.create_random_name(prefix='plan-quick', length=24)
         self.cmd('appservice plan create -g {} -n {}'.format(resource_group, plan))
-        r = self.cmd('webapp create -g {} -n {} --plan {} --deployment-local-git -r "node|20LTS"'.format(
+        r = self.cmd('webapp create -g {} -n {} --plan {} --deployment-local-git -r "node|22LTS"'.format(
             resource_group, webapp_name, plan)).get_output_in_json()
         self.assertTrue(r['ftpPublishingUrl'].startswith('ftps://'))
         self.cmd('webapp config appsettings list -g {} -n {}'.format(resource_group, webapp_name), checks=[
@@ -178,7 +178,7 @@ def test_win_webapp_quick_create_cd(self, resource_group):
         webapp_name = self.create_random_name(prefix='webapp-quick-cd', length=24)
         plan = self.create_random_name(prefix='plan-quick', length=24)
         self.cmd('appservice plan create -g {} -n {}'.format(resource_group, plan))
-        self.cmd('webapp create -g {} -n {} --plan {} --deployment-source-url {} -r "node|20LTS"'.format(
+        self.cmd('webapp create -g {} -n {} --plan {} --deployment-source-url {} -r "node|22LTS"'.format(
             resource_group, webapp_name, plan, TEST_REPO_URL))
         # 30 seconds should be enough for the deployment finished(Skipped under playback mode)
         time.sleep(30)
@@ -231,7 +231,7 @@ def test_linux_webapp_quick_create_cd(self, resource_group):
         plan = 'plan-quick-linux-cd'
         self.cmd(
             'appservice plan create -g {} -n {} --is-linux'.format(resource_group, plan))
-        self.cmd('webapp create -g {} -n {} --plan {} -u {} -r "NODE|20-lts"'.format(resource_group, webapp_name,
+        self.cmd('webapp create -g {} -n {} --plan {} -u {} -r "NODE|22-lts"'.format(resource_group, webapp_name,
                                                                                     plan, TEST_REPO_URL))
         # 45 seconds should be enough for the deployment finished(Skipped under playback mode)
         time.sleep(45)
@@ -486,7 +486,7 @@ def test_download_win_web_log(self, resource_group):
             prefix='webapp-win-log', length=24)
         plan = self.create_random_name(prefix='win-log', length=24)
         self.cmd(f'appservice plan create -g {resource_group} -n {plan} -l eastus')
-        self.cmd('webapp create -g {} -n {} --plan {} --deployment-source-url {} -r "node|20LTS"'.format(
+        self.cmd('webapp create -g {} -n {} --plan {} --deployment-source-url {} -r "node|22LTS"'.format(
             resource_group, webapp_name, plan, TEST_REPO_URL))
         # 30 seconds should be enough for the deployment finished(Skipped under playback mode)
         time.sleep(30)
@@ -808,7 +808,7 @@ def test_webapp_config(self, resource_group):
                      JMESPathCheck('[0].name', '{0}.azurewebsites.net'.format(webapp_name))])
 
         # site azure storage account configurations tests
-        runtime = 'NODE|20-lts'
+        runtime = 'NODE|22-lts'
         linux_plan = self.create_random_name(
             prefix='webapp-linux-plan', length=24)
         linux_webapp = self.create_random_name(
@@ -919,7 +919,7 @@ def test_webapp_config(self, resource_group):
             JMESPathCheck("javaContainer", None),
             JMESPathCheck("javaContainerVersion", None),         
         ])
-        self.cmd('webapp config set -g {} -n {} --runtime NODE:20LTS'.format(resource_group, webapp_name))
+        self.cmd('webapp config set -g {} -n {} --runtime NODE:22LTS'.format(resource_group, webapp_name))
         self.cmd('webapp config appsettings list -g {} -n {}'.format(resource_group, webapp_name)).assert_with_checks([
             JMESPathCheck("[?name=='WEBSITE_NODE_DEFAULT_VERSION']|[0].value", "~20"),
         ])
@@ -948,7 +948,7 @@ def test_webapp_config(self, resource_group):
         self.cmd('webapp config show -g {} -n {}'.format(resource_group, linux_webapp)).assert_with_checks([
             JMESPathCheck("linuxFxVersion", "DOTNETCORE|8.0"),
         ])        
-        self.cmd('webapp config set -g {} -n {} --runtime NODE:20-lts'.format(resource_group, linux_webapp)).assert_with_checks([
+        self.cmd('webapp config set -g {} -n {} --runtime NODE:22-lts'.format(resource_group, linux_webapp)).assert_with_checks([
             JMESPathCheck("linuxFxVersion", "NODE|20-lts"),
         ])
         self.cmd('webapp config show -g {} -n {}'.format(resource_group, linux_webapp)).assert_with_checks([
@@ -1262,7 +1262,7 @@ class LinuxWebappScenarioTest(ScenarioTest):
     @AllowLargeResponse()
     @ResourceGroupPreparer(location=LINUX_ASP_LOCATION_WEBAPP)
     def test_linux_webapp(self, resource_group):
-        runtime = 'NODE|20-lts'
+        runtime = 'NODE|22-lts'
         plan = self.create_random_name(prefix='webapp-linux-plan', length=24)
         webapp = self.create_random_name(prefix='webapp-linux', length=24)
         self.cmd('appservice plan create -g {} -n {} --sku S1 --is-linux' .format(resource_group, plan), checks=[
@@ -1367,7 +1367,7 @@ class LinuxWebappRemoteSSHScenarioTest(ScenarioTest):
     @AllowLargeResponse()
     @ResourceGroupPreparer(location=LINUX_ASP_LOCATION_WEBAPP)
     def test_linux_webapp_remote_ssh(self, resource_group):
-        runtime = 'node|20-lts'
+        runtime = 'node|22-lts'
         plan = self.create_random_name(
             prefix='webapp-remote-ssh-plan', length=40)
         webapp = self.create_random_name(prefix='webapp-remote-ssh', length=40)
@@ -1430,7 +1430,7 @@ class WebappACRScenarioTest(ScenarioTest):
     def test_acr_integration(self, resource_group):
         plan = self.create_random_name(prefix='acrtestplan', length=24)
         webapp = self.create_random_name(prefix='webappacrtest', length=24)
-        runtime = 'NODE|20-lts'
+        runtime = 'NODE|22-lts'
         acr_registry_name = webapp
         self.cmd('acr create --admin-enabled -g {} -n {} --sku Basic'.format(
             resource_group, acr_registry_name))
@@ -2597,7 +2597,7 @@ def test_webapp_linux_acr_use_identity(self, resource_group):
         plan = self.create_random_name(prefix='plan-linux', length=24)
 
         self.cmd('appservice plan create -g {} -n {} --is-linux'.format(resource_group, plan))
-        self.cmd('webapp create -g {} -n {} --plan {} -u {} -r "NODE|20-lts" --assign-identity [system] --acr-use-identity --acr-identity [system]'
+        self.cmd('webapp create -g {} -n {} --plan {} -u {} -r "NODE|22-lts" --assign-identity [system] --acr-use-identity --acr-identity [system]'
                  .format(resource_group, system_identity_webapp_name, plan, TEST_REPO_URL), checks=[
             JMESPathCheck('state', 'Running'),
             JMESPathCheck('name', system_identity_webapp_name),
@@ -2614,7 +2614,7 @@ def test_webapp_linux_acr_use_identity(self, resource_group):
         identity_name = self.create_random_name('app-linux', 16)
         msi_result = self.cmd('identity create -g {} -n {}'.format(
             resource_group, identity_name)).get_output_in_json()
-        self.cmd('webapp create -g {} -n {} --plan {} -u {} -r "NODE|20-lts" --assign-identity {} --acr-use-identity --acr-identity {}'
+        self.cmd('webapp create -g {} -n {} --plan {} -u {} -r "NODE|22-lts" --assign-identity {} --acr-use-identity --acr-identity {}'
                  .format(resource_group, user_identity_webapp_name, plan, TEST_REPO_URL, msi_result['id'], msi_result['id']), checks=[
             JMESPathCheck('state', 'Running'),
             JMESPathCheck('name', user_identity_webapp_name),
@@ -3314,7 +3314,7 @@ def test_webapp_track_runtimestatus_buildfailed(self, resource_group):
         self.cmd(
             'appservice plan create -g {} -n {} --sku S1 --is-linux'.format(resource_group, plan_name))
         self.cmd(
-            'webapp create -g {} -n {} --plan {} -r "NODE|20-LTS"'.format(resource_group, webapp_name, plan_name))
+            'webapp create -g {} -n {} --plan {} -r "NODE|22-LTS"'.format(resource_group, webapp_name, plan_name))
         with self.assertRaisesRegex(CLIError, "Deployment failed because the build process failed"):
             self.cmd('webapp deploy -g {} --n {} --src-path "{}" --type zip --async'.format(resource_group, webapp_name, zip_file))
 
@@ -3327,7 +3327,7 @@ def test_webapp_track_runtimestatus_runtimefailed(self, resource_group):
         self.cmd(
             'appservice plan create -g {} -n {} --sku S1 --is-linux'.format(resource_group, plan_name))
         self.cmd(
-            'webapp create -g {} -n {} --plan {} -r "NODE|20-LTS"'.format(resource_group, webapp_name, plan_name))
+            'webapp create -g {} -n {} --plan {} -r "NODE|22-LTS"'.format(resource_group, webapp_name, plan_name))
         with self.assertRaisesRegex(CLIError, "Deployment failed because the site failed to start within 10 mins."):
             self.cmd('webapp deploy -g {} --n {} --src-path "{}" --type zip --async'.format(resource_group, webapp_name, zip_file))
 
@@ -3375,7 +3375,7 @@ def test_webapp_deployment_source_track_runtimestatus_buildfailed(self, resource
         self.cmd(
             'appservice plan create -g {} -n {} --sku S1 --is-linux'.format(resource_group, plan_name))
         self.cmd(
-            'webapp create -g {} -n {} --plan {} -r "NODE|20-LTS"'.format(resource_group, webapp_name, plan_name))
+            'webapp create -g {} -n {} --plan {} -r "NODE|22-LTS"'.format(resource_group, webapp_name, plan_name))
         with self.assertRaisesRegex(CLIError, "Deployment failed because the build process failed"):
             self.cmd('webapp deployment source config-zip -g {} --n {} --src "{}"'.format(resource_group, webapp_name, zip_file))
 
@@ -3388,7 +3388,7 @@ def test_webapp_deployment_source_track_runtimestatus_runtimefailed(self, resour
         self.cmd(
             'appservice plan create -g {} -n {} --sku S1 --is-linux'.format(resource_group, plan_name))
         self.cmd(
-            'webapp create -g {} -n {} --plan {} -r "NODE|20-LTS"'.format(resource_group, webapp_name, plan_name))
+            'webapp create -g {} -n {} --plan {} -r "NODE|22-LTS"'.format(resource_group, webapp_name, plan_name))
         with self.assertRaisesRegex(CLIError, "Deployment failed because the site failed to start within 10 mins."):
             self.cmd('webapp deployment source config-zip -g {} --n {} --src "{}"'.format(resource_group, webapp_name, zip_file))