diff --git a/linodecli/plugins/get-kubeconfig.py b/linodecli/plugins/get-kubeconfig.py index a839416eb..b75ab3c2b 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,9 +153,26 @@ 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: - yaml.dump(data, file_descriptor) + Path.mkdir(filepath.parent, mode=KUBECONFIG_DIR_MODE, 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. + file_descriptor = os.open( + filepath, + os.O_WRONLY | os.O_CREAT | os.O_TRUNC, + KUBECONFIG_FILE_MODE, + ) + + # 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).st_mode & 0o077: + os.fchmod(file_descriptor, KUBECONFIG_FILE_MODE) + + with os.fdopen(file_descriptor, "w", encoding="utf-8") as file: + yaml.dump(data, file) def _merge_dict(dict_1, dict_2): diff --git a/tests/unit/test_plugin_kubeconfig.py b/tests/unit/test_plugin_kubeconfig.py index ff192f57b..ba2663cdb 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", "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