Tighten history file permissions

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

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

awscli/customizations/history/__init__.py

@@ -10,47 +10,56 @@
 # 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 logging
 import os
 import sys
-import logging
 
-from botocore.history import get_global_history_recorder
 from botocore.exceptions import ProfileNotFound
+from botocore.history import get_global_history_recorder
 
 from awscli.compat import sqlite3
 from awscli.customizations.commands import BasicCommand
-from awscli.customizations.history.constants import HISTORY_FILENAME_ENV_VAR
-from awscli.customizations.history.constants import DEFAULT_HISTORY_FILENAME
-from awscli.customizations.history.db import DatabaseConnection
-from awscli.customizations.history.db import DatabaseRecordWriter
-from awscli.customizations.history.db import RecordBuilder
-from awscli.customizations.history.db import DatabaseHistoryHandler
-from awscli.customizations.history.show import ShowCommand
+from awscli.customizations.history.constants import (
+    DEFAULT_HISTORY_FILENAME,
+    HISTORY_FILENAME_ENV_VAR,
+)
+from awscli.customizations.history.db import (
+    DatabaseConnection,
+    DatabaseHistoryHandler,
+    DatabaseRecordWriter,
+    RecordBuilder,
+)
 from awscli.customizations.history.list import ListCommand
-
+from awscli.customizations.history.show import ShowCommand
 
 LOG = logging.getLogger(__name__)
 HISTORY_RECORDER = get_global_history_recorder()
 
 
 def register_history_mode(event_handlers):
-    event_handlers.register(
-        'session-initialized', attach_history_handler)
+    event_handlers.register('session-initialized', attach_history_handler)
 
 
 def register_history_commands(event_handlers):
     event_handlers.register(
-        "building-command-table.main", add_history_commands)
+        "building-command-table.main", add_history_commands
+    )
 
 
 def attach_history_handler(session, parsed_args, **kwargs):
     if _should_enable_cli_history(session, parsed_args):
         LOG.debug('Enabling CLI history')
 
         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_FILENAME_ENV_VAR, DEFAULT_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)
@@ -98,10 +107,12 @@ class HistoryCommand(BasicCommand):
     )
     SUBCOMMANDS = [
         {'name': 'show', 'command_class': ShowCommand},
-        {'name': 'list', 'command_class': ListCommand}
+        {'name': 'list', 'command_class': ListCommand},
     ]
 
     def _run_main(self, parsed_args, parsed_globals):
         if parsed_args.subcommand is None:
-            raise ValueError("usage: aws [options] <command> <subcommand> "
-                             "[parameters]\naws: error: too few arguments")
+            raise ValueError(
+                "usage: aws [options] <command> <subcommand> "
+                "[parameters]\naws: error: too few arguments"
+            )

awscli/customizations/history/db.py

@@ -10,24 +10,22 @@
 # 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 uuid
-import time
-import json
 import datetime
-import threading
+import json
 import logging
-from awscli.compat import collections_abc
+import os
+import threading
+import time
+import uuid
 
 from botocore.history import BaseHistoryHandler
 
-from awscli.compat import sqlite3
-from awscli.compat import binary_type
-
+from awscli.compat import binary_type, collections_abc, sqlite3
 
 LOG = logging.getLogger(__name__)
 
 
-class DatabaseConnection(object):
+class DatabaseConnection:
     _CREATE_TABLE = """
         CREATE TABLE IF NOT EXISTS records (
           id TEXT,
@@ -40,8 +38,18 @@ class DatabaseConnection(object):
     _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)
+            db_filename, check_same_thread=False, isolation_level=None
+        )
         self._ensure_database_setup()
 
     def close(self):
@@ -92,8 +100,9 @@ def _remove_non_unicode_stings(self, obj):
         if isinstance(obj, str):
             obj = self._try_decode_bytes(obj)
         elif isinstance(obj, dict):
-            obj = dict((k, self._remove_non_unicode_stings(v)) for k, v
-                       in obj.items())
+            obj = dict(
+                (k, self._remove_non_unicode_stings(v)) for k, v in obj.items()
+            )
         elif isinstance(obj, (list, tuple)):
             obj = [self._remove_non_unicode_stings(o) for o in obj]
         return obj
@@ -132,7 +141,7 @@ def default(self, obj):
             return repr(obj)
 
 
-class DatabaseRecordWriter(object):
+class DatabaseRecordWriter:
     _WRITE_RECORD = """
         INSERT INTO records(
             id, request_id, source, event_type, timestamp, payload)
@@ -152,26 +161,30 @@ def write_record(self, record):
 
     def _create_db_record(self, record):
         event_type = record['event_type']
-        json_serialized_payload = json.dumps(record['payload'],
-                                             cls=PayloadSerializer)
+        json_serialized_payload = json.dumps(
+            record['payload'], cls=PayloadSerializer
+        )
         db_record = (
             record['command_id'],
             record.get('request_id'),
             record['source'],
             event_type,
             record['timestamp'],
-            json_serialized_payload
+            json_serialized_payload,
         )
         return db_record
 
 
-class DatabaseRecordReader(object):
+class DatabaseRecordReader:
     _ORDERING = 'ORDER BY timestamp'
-    _GET_LAST_ID_RECORDS = """
+    _GET_LAST_ID_RECORDS = (
+        """
         SELECT * FROM records
         WHERE id =
         (SELECT id FROM records WHERE timestamp =
-        (SELECT max(timestamp) FROM records)) %s;""" % _ORDERING
+        (SELECT max(timestamp) FROM records)) %s;"""
+        % _ORDERING
+    )
     _GET_RECORDS_BY_ID = 'SELECT * from records where id = ? %s' % _ORDERING
     _GET_ALL_RECORDS = (
         'SELECT a.id AS id_a, '
@@ -218,9 +231,10 @@ def iter_all_records(self):
             yield row
 
 
-class RecordBuilder(object):
+class RecordBuilder:
     _REQUEST_LIFECYCLE_EVENTS = set(
-        ['API_CALL', 'HTTP_REQUEST', 'HTTP_RESPONSE', 'PARSED_RESPONSE'])
+        ['API_CALL', 'HTTP_REQUEST', 'HTTP_RESPONSE', 'PARSED_RESPONSE']
+    )
     _START_OF_REQUEST_LIFECYCLE_EVENT = 'API_CALL'
 
     def __init__(self):
@@ -254,7 +268,7 @@ def build_record(self, event_type, payload, source):
             'event_type': event_type,
             'payload': payload,
             'source': source,
-            'timestamp': int(time.time() * 1000)
+            'timestamp': int(time.time() * 1000),
         }
         request_id = self._get_request_id(event_type)
         if request_id: