diff --git a/linodecli/plugins/get-kubeconfig.py b/linodecli/plugins/get-kubeconfig.py index a839416eb..c0204ba71 100644 --- a/linodecli/plugins/get-kubeconfig.py +++ b/linodecli/plugins/get-kubeconfig.py @@ -8,6 +8,7 @@ import argparse import base64 +import os import sys from pathlib import Path @@ -19,6 +20,11 @@ PLUGIN_BASE = "linode-cli get-kubeconfig" +# Kubeconfigs contain credentials, so they should only be +# accessible by the user that created them. +KUBECONFIG_FILE_MODE = 0o600 +KUBECONFIG_DIR_MODE = 0o700 + def call(args, context): """ @@ -147,8 +153,27 @@ def _load_config(filepath): # Dumps data to a yaml file def _dump_config(filepath, data): - Path.mkdir(filepath.parent, exist_ok=True) - with open(filepath, "w", encoding="utf-8") as file_descriptor: + filepath.parent.mkdir(mode=KUBECONFIG_DIR_MODE, parents=True, exist_ok=True) + + # Create the file with restrictive permissions rather than chmod-ing it + # afterwards, so its contents are never briefly readable by other users. + # NOTE: The mode is only applied when the file is created. + def opener(path, flags): + return os.open(path, flags, mode=KUBECONFIG_FILE_MODE) + + with open( + filepath, "w", encoding="utf-8", opener=opener + ) as file_descriptor: + # Tighten the permissions of pre-existing files that are readable or + # writable by users other than the owner. + # NOTE: os.fchmod is not available on Windows, where POSIX file modes + # are not meaningful anyway. + if ( + hasattr(os, "fchmod") + and os.fstat(file_descriptor.fileno()).st_mode & 0o077 + ): + os.fchmod(file_descriptor.fileno(), KUBECONFIG_FILE_MODE) + yaml.dump(data, file_descriptor) diff --git a/tests/unit/test_plugin_kubeconfig.py b/tests/unit/test_plugin_kubeconfig.py index ff192f57b..7fdc0b8da 100644 --- a/tests/unit/test_plugin_kubeconfig.py +++ b/tests/unit/test_plugin_kubeconfig.py @@ -204,6 +204,64 @@ def test_merge(mock_cli, fake_kubeconfig_file): assert result["dictionary"] == yaml_a["dictionary"] +# Ensure newly created kubeconfig files are not world/group-readable +@pytest.mark.skipif( + os.name == "nt", reason="POSIX file modes are not supported on Windows" +) +def test_written_config_permissions(mock_cli): + mock_cli.call_operation = mock_call_operation + + with tempfile.TemporaryDirectory() as temp_dir: + file_path = os.path.join(temp_dir, "new_dir", "nested", "config") + + try: + plugin.call( + [ + "--label", + "nonempty_data", + "--kubeconfig", + file_path, + ], + PluginContext("REALTOKEN", mock_cli), + ) + except SystemExit as err: + assert err.code == 0 + + assert os.path.exists(file_path) + assert os.stat(file_path).st_mode & 0o777 == 0o600 + assert os.stat(os.path.dirname(file_path)).st_mode & 0o777 == 0o700 + + +# Ensure pre-existing world-readable kubeconfig files get tightened +@pytest.mark.skipif( + os.name == "nt", reason="POSIX file modes are not supported on Windows" +) +def test_existing_config_permissions_tightened(mock_cli): + mock_cli.call_operation = mock_call_operation + + with tempfile.TemporaryDirectory() as temp_dir: + file_path = os.path.join(temp_dir, "config") + + with open(file_path, "w", encoding="utf-8") as file: + file.write(TEST_YAML_CONTENT_A) + os.chmod(file_path, 0o644) + + try: + plugin.call( + [ + "--label", + "nonempty_data", + "--kubeconfig", + file_path, + ], + PluginContext("REALTOKEN", mock_cli), + ) + except SystemExit as err: + assert err.code == 0 + + assert os.stat(file_path).st_mode & 0o777 == 0o600 + + def test_merge_to_empty_config(mock_cli, fake_kubeconfig_file_without_entries): stdout_buf = io.StringIO() mock_cli.call_operation = mock_call_operation