Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 26 additions & 3 deletions linodecli/plugins/get-kubeconfig.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import argparse
import base64
import os
import sys
from pathlib import Path

Expand All @@ -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):
"""
Expand Down Expand Up @@ -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):
Expand Down
58 changes: 58 additions & 0 deletions tests/unit/test_plugin_kubeconfig.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down