Tighten history file permissions

.changes/next-release/enhancement-clihistory-87056.json

@@ -0,0 +1,5 @@
+{
+  "type": "enhancement",
+  "category": "cli-history",
+  "description": "Create local history files with specific permissions"
+}

awscli/autoprompt/history.py

@@ -44,15 +44,24 @@ def load_history_strings(self):
     def store_string(self, string):
         history = {'version': self.HISTORY_VERSION, 'commands': []}
         try:
+            dir_path = os.path.dirname(self.filename)
             if os.path.exists(self.filename):
                 with open(self.filename) as f:
                     history = json.load(f)
-            elif not os.path.exists(os.path.dirname(self.filename)):
-                os.makedirs(os.path.dirname(self.filename))
+            if not os.path.exists(dir_path):
+                os.makedirs(dir_path)
+            try:
+                os.chmod(dir_path, 0o700)
+            except OSError as e:
+                LOG.debug('Unable to set directory permissions: %s', e)
             history['commands'].append(string)
             history['commands'] = history['commands'][-self._max_commands :]
             with open(self.filename, 'w') as f:
                 json.dump(history, f)
+            try:
+                os.chmod(self.filename, 0o600)
+            except OSError as e:
+                LOG.debug('Unable to set file permissions: %s', e)
         except Exception:
             LOG.debug('Exception on loading prompt history:', exc_info=True)
 

awscli/botocore/utils.py

@@ -3943,6 +3943,10 @@ def __setitem__(self, cache_key, value):
             )
         if not os.path.isdir(self._working_dir):
             os.makedirs(self._working_dir)
+        try:
+            os.chmod(self._working_dir, 0o700)
+        except OSError as e:
+            logger.debug('Unable to set directory permissions: %s', e)
         with os.fdopen(
             os.open(full_key, os.O_WRONLY | os.O_CREAT, 0o600), 'w'
         ) as f:

awscli/customizations/history/__init__.py

@@ -53,8 +53,14 @@ def attach_history_handler(session, parsed_args, **kwargs):
         history_filename = os.environ.get(
             HISTORY_FILENAME_ENV_VAR, DEFAULT_HISTORY_FILENAME
         )
-        if not os.path.isdir(os.path.dirname(history_filename)):
-            os.makedirs(os.path.dirname(history_filename))
+
+        history_dir = os.path.dirname(history_filename)
+        if not os.path.isdir(history_dir):
+            os.makedirs(history_dir)
+        try:
+            os.chmod(history_dir, 0o700)
+        except OSError as e:
+            LOG.debug('Unable to set directory permissions: %s', e)
 
         connection = DatabaseConnection(history_filename)
         writer = DatabaseRecordWriter(connection)

awscli/customizations/history/db.py

@@ -13,6 +13,7 @@
 import datetime
 import json
 import logging
+import os
 import threading
 import time
 import uuid
@@ -37,6 +38,15 @@ class DatabaseConnection:
     _ENABLE_WAL = 'PRAGMA journal_mode=WAL'
 
     def __init__(self, db_filename):
+        # Skip file operations for in-memory databases
+        if db_filename != ':memory:':
+            if not os.path.exists(db_filename):
+                # Create file so we can set permissions before sqlite opens it
+                open(db_filename, 'a').close()
+            try:
+                os.chmod(db_filename, 0o600)
+            except OSError as e:
+                LOG.debug('Unable to set file permissions: %s', e)
         self._connection = sqlite3.connect(
             db_filename, check_same_thread=False, isolation_level=None
         )

awscli/telemetry.py

@@ -11,6 +11,7 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 import io
+import logging
 import os
 import sqlite3
 import sys
@@ -28,6 +29,8 @@
 from awscli.compat import is_windows
 from awscli.utils import add_component_to_user_agent_extra
 
+LOG = logging.getLogger(__name__)
+
 _CACHE_DIR = Path.home() / '.aws' / 'cli' / 'cache'
 _DATABASE_FILENAME = 'session.db'
 _SESSION_LENGTH_SECONDS = 60 * 30
@@ -77,12 +80,15 @@ class CLISessionDatabaseConnection:
     _ENABLE_WAL = 'PRAGMA journal_mode=WAL'
 
     def __init__(self, connection=None):
-        self._connection = connection or sqlite3.connect(
-            _CACHE_DIR / _DATABASE_FILENAME,
-            check_same_thread=False,
-            isolation_level=None,
-        )
-        self._ensure_cache_dir()
+        self._connection = connection
+        if self._connection is None:
+            self._ensure_cache_dir()
+            self._ensure_database_file()
+            self._connection = sqlite3.connect(
+                _CACHE_DIR / _DATABASE_FILENAME,
+                check_same_thread=False,
+                isolation_level=None,
+            )
         self._ensure_database_setup()
 
     def execute(self, query, *parameters):
@@ -96,6 +102,19 @@ def execute(self, query, *parameters):
 
     def _ensure_cache_dir(self):
         _CACHE_DIR.mkdir(parents=True, exist_ok=True)
+        try:
+            os.chmod(_CACHE_DIR, 0o700)
+        except OSError as e:
+            LOG.debug('Unable to set directory permissions: %s', e)
+
+    def _ensure_database_file(self):
+        db_path = _CACHE_DIR / _DATABASE_FILENAME
+        if not db_path.exists():
+            open(db_path, 'a').close()
+        try:
+            os.chmod(db_path, 0o600)
+        except OSError as e:
+            LOG.debug('Unable to set file permissions: %s', e)
 
     def _ensure_database_setup(self):
         self._create_session_table()

tests/functional/test_telemetry.py

@@ -10,7 +10,10 @@
 # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
+import os
 import sqlite3
+import stat
+from pathlib import Path
 from unittest.mock import MagicMock, PropertyMock, patch
 
 import pytest
@@ -27,6 +30,7 @@
     CLISessionOrchestrator,
     add_session_id_component_to_user_agent_extra,
 )
+from awscli.testutils import FileCreator
 from tests.markers import skip_if_windows
 
 
@@ -130,6 +134,71 @@ def execute(self, query, *parameters):
         assert cursor.fetchall() == []
 
 
+@skip_if_windows
+class TestCLISessionDatabaseConnectionPermissions:
+    def setup_method(self):
+        self.files = FileCreator()
+
+    def teardown_method(self):
+        self.files.remove_all()
+
+    def test_create_directory_with_secure_permissions(self):
+        cache_dir = self.files.full_path('cache')
+        cache_path = Path(cache_dir)
+
+        with patch('awscli.telemetry._CACHE_DIR', cache_path):
+            with patch('awscli.telemetry._DATABASE_FILENAME', 'session.db'):
+                conn = CLISessionDatabaseConnection()
+                conn._connection.close()
+
+        assert os.path.isdir(cache_dir)
+        dir_mode = stat.S_IMODE(os.stat(cache_dir).st_mode)
+        assert dir_mode == 0o700
+
+    def test_tighten_existing_directory_permissions(self):
+        cache_dir = self.files.full_path('cache')
+        os.makedirs(cache_dir, mode=0o755)
+        cache_path = Path(cache_dir)
+
+        with patch('awscli.telemetry._CACHE_DIR', cache_path):
+            with patch('awscli.telemetry._DATABASE_FILENAME', 'session.db'):
+                conn = CLISessionDatabaseConnection()
+                conn._connection.close()
+
+        dir_mode = stat.S_IMODE(os.stat(cache_dir).st_mode)
+        assert dir_mode == 0o700
+
+    def test_create_database_file_with_secure_permissions(self):
+        cache_dir = self.files.full_path('cache')
+        db_file = os.path.join(cache_dir, 'session.db')
+        cache_path = Path(cache_dir)
+
+        with patch('awscli.telemetry._CACHE_DIR', cache_path):
+            with patch('awscli.telemetry._DATABASE_FILENAME', 'session.db'):
+                conn = CLISessionDatabaseConnection()
+                conn._connection.close()
+
+        assert os.path.isfile(db_file)
+        file_mode = stat.S_IMODE(os.stat(db_file).st_mode)
+        assert file_mode == 0o600
+
+    def test_tighten_existing_database_file_permissions(self):
+        cache_dir = self.files.full_path('cache')
+        os.makedirs(cache_dir, mode=0o700)
+        db_file = os.path.join(cache_dir, 'session.db')
+        open(db_file, 'a').close()
+        os.chmod(db_file, 0o644)
+        cache_path = Path(cache_dir)
+
+        with patch('awscli.telemetry._CACHE_DIR', cache_path):
+            with patch('awscli.telemetry._DATABASE_FILENAME', 'session.db'):
+                conn = CLISessionDatabaseConnection()
+                conn._connection.close()
+
+        file_mode = stat.S_IMODE(os.stat(db_file).st_mode)
+        assert file_mode == 0o600
+
+
 class TestCLISessionDatabaseWriter:
     def test_write(self, session_writer, session_reader, session_sweeper):
         session_writer.write(

tests/unit/autoprompt/test_history.py

@@ -13,8 +13,10 @@
 import json
 import os
 import shutil
+import stat
 import tempfile
 
+import pytest
 from prompt_toolkit.buffer import Buffer
 from prompt_toolkit.completion import Completion
 from prompt_toolkit.document import Document
@@ -23,6 +25,7 @@
 
 from awscli.autoprompt.history import HistoryCompleter, HistoryDriver
 from awscli.testutils import mock, unittest
+from tests.markers import skip_if_windows
 
 
 class TestHistoryCompleter(unittest.TestCase):
@@ -190,3 +193,60 @@ def test_handle_io_errors(self, file_history_mock):
         history_driver = HistoryDriver(self.filename)
         file_history_mock.store_string.side_effect = IOError
         history_driver.store_string('aws dynamodb create-table')
+
+
+@pytest.fixture
+def temp_dir():
+    dirname = tempfile.mkdtemp()
+    yield dirname
+    shutil.rmtree(dirname)
+
+
+@skip_if_windows
+def test_create_directory_with_secure_permissions(temp_dir):
+    subdir = os.path.join(temp_dir, 'newdir')
+    filename = os.path.join(subdir, 'prompt_history.json')
+    history_driver = HistoryDriver(filename)
+    history_driver.store_string('aws ec2 describe-instances')
+
+    assert os.path.isdir(subdir)
+    dir_mode = stat.S_IMODE(os.stat(subdir).st_mode)
+    assert dir_mode == 0o700
+
+
+@skip_if_windows
+def test_tighten_existing_directory_permissions(temp_dir):
+    subdir = os.path.join(temp_dir, 'existingdir')
+    os.makedirs(subdir, mode=0o755)
+    filename = os.path.join(subdir, 'prompt_history.json')
+
+    history_driver = HistoryDriver(filename)
+    history_driver.store_string('aws ec2 describe-instances')
+
+    dir_mode = stat.S_IMODE(os.stat(subdir).st_mode)
+    assert dir_mode == 0o700
+
+
+@skip_if_windows
+def test_create_file_with_secure_permissions(temp_dir):
+    filename = os.path.join(temp_dir, 'prompt_history.json')
+    history_driver = HistoryDriver(filename)
+    history_driver.store_string('aws ec2 describe-instances')
+
+    assert os.path.isfile(filename)
+    file_mode = stat.S_IMODE(os.stat(filename).st_mode)
+    assert file_mode == 0o600
+
+
+@skip_if_windows
+def test_tighten_existing_file_permissions(temp_dir):
+    filename = os.path.join(temp_dir, 'prompt_history.json')
+    with open(filename, 'w') as f:
+        json.dump({'version': 1, 'commands': []}, f)
+    os.chmod(filename, 0o644)
+
+    history_driver = HistoryDriver(filename)
+    history_driver.store_string('aws ec2 describe-instances')
+
+    file_mode = stat.S_IMODE(os.stat(filename).st_mode)
+    assert file_mode == 0o600

tests/unit/customizations/history/test_db.py

@@ -15,6 +15,7 @@
 import numbers
 import os
 import re
+import stat
 import threading
 
 from awscli.compat import queue
@@ -28,6 +29,7 @@
 )
 from awscli.testutils import FileCreator, mock, unittest
 from tests import CaseInsensitiveDict
+from tests.markers import skip_if_windows
 
 
 class FakeDatabaseConnection:
@@ -48,8 +50,20 @@ def tearDown(self):
 
 
 class TestDatabaseConnection(unittest.TestCase):
+    def setUp(self):
+        self.files = FileCreator()
+
+    def tearDown(self):
+        self.files.remove_all()
+
+    @skip_if_windows
+    @mock.patch('awscli.customizations.history.db.os.chmod')
+    @mock.patch('awscli.customizations.history.db.os.path.exists')
     @mock.patch('awscli.compat.sqlite3.connect')
-    def test_can_connect_to_argument_file(self, mock_connect):
+    def test_can_connect_to_argument_file(
+        self, mock_connect, mock_exists, mock_chmod
+    ):
+        mock_exists.return_value = True
         expected_location = os.path.expanduser(
             os.path.join('~', 'foo', 'bar', 'baz.db')
         )
@@ -64,7 +78,7 @@ def test_does_try_to_enable_wal(self, mock_connect):
         conn._connection.execute.assert_any_call('PRAGMA journal_mode=WAL')
 
     def test_does_ensure_table_created_first(self):
-        db = DatabaseConnection(":memory:")
+        db = DatabaseConnection(':memory:')
         cursor = db.execute('PRAGMA table_info(records)')
         schema = [col[:3] for col in cursor.fetchall()]
         expected_schema = [
@@ -85,6 +99,23 @@ def test_can_close(self, mock_connect):
         conn.close()
         self.assertTrue(connection.close.called)
 
+    @skip_if_windows
+    def test_create_new_file_with_secure_permissions(self):
+        db_filename = os.path.join(self.files.rootdir, 'new_secure.db')
+        DatabaseConnection(db_filename)
+        file_mode = stat.S_IMODE(os.stat(db_filename).st_mode)
+        self.assertEqual(file_mode, 0o600)
+
+    @skip_if_windows
+    def test_tighten_existing_file_permissions(self):
+        db_filename = os.path.join(self.files.rootdir, 'existing.db')
+        open(db_filename, 'a').close()
+        os.chmod(db_filename, 0o644)
+
+        DatabaseConnection(db_filename)
+        file_mode = stat.S_IMODE(os.stat(db_filename).st_mode)
+        self.assertEqual(file_mode, 0o600)
+
 
 class TestDatabaseHistoryHandler(unittest.TestCase):
     UUID_PATTERN = re.compile(