Create private key files with read and write permissions for the user only

securesystemslib/interface.py

@@ -54,9 +54,6 @@
 DEFAULT_RSA_KEY_BITS = 3072
 
 
-
-
-
 def get_password(prompt='Password: ', confirm=False):
   """Prompts user to enter a password.
 
@@ -204,6 +201,7 @@ def _generate_and_write_rsa_keypair(filepath=None, bits=DEFAULT_RSA_KEY_BITS,
   Side Effects:
     Prompts user for a password if 'prompt' is True.
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -239,7 +237,7 @@ def _generate_and_write_rsa_keypair(filepath=None, bits=DEFAULT_RSA_KEY_BITS,
   # Write PEM-encoded private key to <filepath>
   file_object = tempfile.TemporaryFile()
   file_object.write(private.encode('utf-8'))
-  util.persist_temp_file(file_object, filepath)
+  util.persist_temp_file(file_object, filepath, restrict=True)
 
   return filepath
 
@@ -270,6 +268,7 @@ def generate_and_write_rsa_keypair(password, filepath=None,
 
   Side Effects:
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -306,6 +305,7 @@ def generate_and_write_rsa_keypair_with_prompt(filepath=None,
   Side Effects:
     Prompts user for a password.
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -338,6 +338,7 @@ def generate_and_write_unencrypted_rsa_keypair(filepath=None,
 
   Side Effects:
     Writes unencrypted key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -469,6 +470,7 @@ def _generate_and_write_ed25519_keypair(filepath=None, password=None,
   Side Effects:
     Prompts user for a password if 'prompt' is True.
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -508,7 +510,7 @@ def _generate_and_write_ed25519_keypair(filepath=None, password=None,
   # Write private key to <filepath>
   file_object = tempfile.TemporaryFile()
   file_object.write(ed25519_key.encode('utf-8'))
-  util.persist_temp_file(file_object, filepath)
+  util.persist_temp_file(file_object, filepath, restrict=True)
 
   return filepath
 
@@ -536,6 +538,7 @@ def generate_and_write_ed25519_keypair(password, filepath=None):
 
   Side Effects:
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -568,6 +571,7 @@ def generate_and_write_ed25519_keypair_with_prompt(filepath=None):
   Side Effects:
     Prompts user for a password.
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -595,6 +599,7 @@ def generate_and_write_unencrypted_ed25519_keypair(filepath=None):
 
   Side Effects:
     Writes unencrypted key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -713,6 +718,7 @@ def _generate_and_write_ecdsa_keypair(filepath=None, password=None,
   Side Effects:
     Prompts user for a password if 'prompt' is True.
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -752,7 +758,7 @@ def _generate_and_write_ecdsa_keypair(filepath=None, password=None,
   # Write private key to <filepath>
   file_object = tempfile.TemporaryFile()
   file_object.write(ecdsa_key.encode('utf-8'))
-  util.persist_temp_file(file_object, filepath)
+  util.persist_temp_file(file_object, filepath, restrict=True)
 
   return filepath
 
@@ -780,6 +786,7 @@ def generate_and_write_ecdsa_keypair(password, filepath=None):
 
   Side Effects:
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -812,6 +819,7 @@ def generate_and_write_ecdsa_keypair_with_prompt(filepath=None):
   Side Effects:
     Prompts user for a password.
     Writes key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.
@@ -839,6 +847,7 @@ def generate_and_write_unencrypted_ecdsa_keypair(filepath=None):
 
   Side Effects:
     Writes unencrypted key files to disk.
+    Overwrites files if they already exist.
 
   Returns:
     The private key filepath.

securesystemslib/storage.py

@@ -20,9 +20,10 @@
 import logging
 import os
 import shutil
+import stat
 from contextlib import contextmanager
 from securesystemslib import exceptions
-from typing import BinaryIO, IO, Iterator, List
+from typing import BinaryIO, IO, Iterator, List, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -65,7 +66,8 @@ def get(self, filepath: str) -> Iterator[BinaryIO]:
 
 
   @abc.abstractmethod
-  def put(self, fileobj: IO, filepath: str) -> None:
+  def put(self, fileobj: IO, filepath: str, restrict: Optional[bool] = False
+        ) -> None:
     """
     <Purpose>
       Store a file-like object in the storage backend.
@@ -79,6 +81,13 @@ def put(self, fileobj: IO, filepath: str) -> None:
       filepath:
         The full path to the location where 'fileobj' will be stored.
 
+      restrict:
+        Whether the file should be created with restricted permissions.
+        What counts as restricted is backend-specific. For a filesystem on a
+        UNIX-like operating system, that may mean read/write permissions only
+        for the user (octal mode 0o600). For a cloud storage system, that
+        likely means Cloud provider specific ACL restrictions.
+
     <Exceptions>
       securesystemslib.exceptions.StorageError, if the file can not be stored.
 
@@ -208,14 +217,35 @@ def get(self, filepath:str) -> Iterator[BinaryIO]:
         file_object.close()
 
 
-  def put(self, fileobj: IO, filepath: str) -> None:
+  def put(self, fileobj: IO, filepath: str, restrict: Optional[bool] = False
+        ) -> None:
     # If we are passed an open file, seek to the beginning such that we are
     # copying the entire contents
     if not fileobj.closed:
       fileobj.seek(0)
 
+    # If a file with the same name already exists, the new permissions
+    # may not be applied.
+    try:
+      os.remove(filepath)
+    except OSError:
+      pass
+
     try:
-      with open(filepath, 'wb') as destination_file:
+      if restrict:
+        # On UNIX-based systems restricted files are created with read and
+        # write permissions for the user only (octal value 0o600).
+        fd = os.open(filepath, os.O_WRONLY|os.O_CREAT,
+          stat.S_IRUSR|stat.S_IWUSR)
+      else:
+        # Non-restricted files use the default 'mode' argument of os.open()
+        # granting read, write, and execute for all users (octal mode 0o777).
+        # NOTE: mode may be modified by the user's file mode creation mask
+        # (umask) or on Windows limited to the smaller set of OS supported
+        # permisssions.
+        fd = os.open(filepath, os.O_WRONLY|os.O_CREAT)
+
+      with os.fdopen(fd, "wb") as destination_file:
         shutil.copyfileobj(fileobj, destination_file)
         # Force the destination file to be written to disk from Python's internal
         # and the operating system's buffers.  os.fsync() should follow flush().

securesystemslib/util.py

@@ -179,7 +179,8 @@ def persist_temp_file(
     temp_file: IO,
     persist_path: str,
     storage_backend: Optional[StorageBackendInterface] = None,
-    should_close: bool = True
+    should_close: bool = True,
+    restrict: bool = False
 ) -> None:
   """
   <Purpose>
@@ -203,6 +204,11 @@ def persist_temp_file(
       A boolean indicating whether the file should be closed after it has been
       persisted. Default is True, the file is closed.
 
+    restrict:
+      A boolean indicating whether the file should have restricted privileges.
+      What evactly counts as restricted privileges is an implementation detail
+      of the backing StorageBackendInterface implementation.
+
   <Exceptions>
     securesystemslib.exceptions.StorageError: If file cannot be written.
 
@@ -213,7 +219,7 @@ def persist_temp_file(
   if storage_backend is None:
     storage_backend = FilesystemBackend()
 
-  storage_backend.put(temp_file, persist_path)
+  storage_backend.put(temp_file, persist_path, restrict=restrict)
 
   if should_close:
     temp_file.close()

tests/test_interface.py

@@ -633,6 +633,7 @@ def test_generate_keypair_wrappers(self):
 
     """
     key_pw = "pw"
+    expected_priv_mode = stat.S_IFREG|stat.S_IRUSR|stat.S_IWUSR
     for idx, (gen, gen_prompt, gen_plain, import_priv, schema) in enumerate([
         (
           generate_and_write_rsa_keypair,
@@ -661,6 +662,10 @@ def test_generate_keypair_wrappers(self):
       priv = import_priv(fn_encrypted, key_pw)
       self.assertTrue(schema.matches(priv), assert_msg)
 
+      # Test that encrypted private key is generated with read and write
+      # permissions for user only
+      self.assertEqual(os.stat(fn_encrypted).st_mode, expected_priv_mode)
+
       # Test generate_and_write_*_keypair errors if password is None or empty
       with self.assertRaises(FormatError, msg=assert_msg):
         fn_encrypted = gen(None)
@@ -688,6 +693,9 @@ def test_generate_keypair_wrappers(self):
       priv = import_priv(fn_unencrypted)
       self.assertTrue(schema.matches(priv), assert_msg)
 
+      # Test that unencrypted private key is generated with read and write
+      # permissions for user only
+      self.assertEqual(os.stat(fn_unencrypted).st_mode, expected_priv_mode)
 
 
   def test_import_publickeys_from_file(self):

tests/test_util.py

@@ -24,6 +24,7 @@
 import tempfile
 import unittest
 import timeit
+import stat
 
 import securesystemslib.settings
 import securesystemslib.hash
@@ -240,8 +241,18 @@ def test_B9_persist_temp_file(self):
     dest_path = os.path.join(dest_temp_dir, self.random_string())
     tmpfile = tempfile.TemporaryFile()
     tmpfile.write(self.random_string().encode('utf-8'))
-    securesystemslib.util.persist_temp_file(tmpfile, dest_path)
+
+    # Write a file with restricted permissions
+    securesystemslib.util.persist_temp_file(tmpfile, dest_path, restrict=True)
     self.assertTrue(dest_path)
+
+    # Need to set also the stat.S_IFREG bit to match the st_mode output
+    # stat.S_IFREG - Regular file
+    expected_mode = stat.S_IFREG | stat.S_IRUSR | stat.S_IWUSR
+    if os.name == 'nt':
+      # Windows only supports setting the read-only attribute.
+      expected_mode = stat.S_IFREG | stat.S_IWUSR | stat.S_IRUSR | stat.S_IWGRP | stat.S_IRGRP | stat.S_IWOTH | stat.S_IROTH
+    self.assertEqual(os.stat(dest_path).st_mode, expected_mode)
     self.assertTrue(tmpfile.closed)
 
     # Test persisting a file without automatically closing the tmpfile