【Python中級】PythonでLinuxコマンドをセキュアに実行

Python中級

PythonでLinuxのアプリを作っていると、管理者以外アクセスできないファイルへアクセスしたいときが出てくる。例えば/root/.に隠した秘密鍵ファイルにアクセスしたいときだ。こういったときに使うセキュアなコードを備忘として書き残しておく。

1.構成

Linuxへsudoで実行コマンドを渡す関数と、それにファイルの読み書き・作成・削除コマンドを送る関数で構成する。また、rootとsudoユーザーによって実行コマンドを分けるようにする。

実験用のファイル構成は以下の通り。

/root/.test/
 └─ pass.txt

2.実装

#!/usr/bin/env python3
"""
root 専用パスフレーズファイル(PASS_FILE)を安全に読み書きするテストツール。
- PASS_DIR: /root/.test
- PASS_FILE: /root/.test/pass.txt

非 root で実行した場合は sudo 経由で root のファイル・ディレクトリを操作し、
root で実行した場合はローカルの os / shutil で直接操作する。
"""

from __future__ import annotations
import os
import sys
import shlex
import subprocess
import shutil
from typing import Optional

PASS_DIR = "/root/.test"
PASS_FILE = os.path.join(PASS_DIR, "pass.txt")


# ======================================================================
# 共通ユーティリティ
# ======================================================================

def is_root() -> bool:
    """現在のプロセスが root かどうかを判定する。"""
    return os.geteuid() == 0


def run_sudo_cmd(
    cmd: list[str],
    input_data: Optional[bytes] = None,
) -> subprocess.CompletedProcess:
    """
    sudo 経由でコマンドを実行する共通関数。

    Parameters
    ----------
    cmd : list[str]
        実行したいコマンド(["mkdir", "-p", "/path"] のような形)。
    input_data : Optional[bytes]
        標準入力に流し込むバイト列(cat > file などのときに使う)。

    Returns
    -------
    subprocess.CompletedProcess
        実行結果。returncode / stdout / stderr を参照する。
    """
    full = ["sudo"] + cmd
    return subprocess.run(
        full,
        input=input_data,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )


# ======================================================================
# sudo 経由のファイル/ディレクトリ操作
# ======================================================================

def sudo_file_write(path: str, data: bytes, mode: int = 0o600) -> None:
    """
    sudo 経由でファイルを書き込む(上書き)。

    - umask 077 で cat > file することで、書き込み時に 600 に近い権限で作成。
    - その後 chmod で mode を明示的に設定する。
    - shlex.quote により path をシェル用にエスケープして安全性を確保。
    """
    safe_path = shlex.quote(path)
    proc = run_sudo_cmd(
        ["bash", "-c", f"umask 077; cat > {safe_path}"],
        input_data=data,
    )
    if proc.returncode != 0:
        raise RuntimeError(
            f"sudo write failed: {proc.stderr.decode(errors='ignore')}"
        )

    # 明示的に chmod しておく(mode は 0o600 などの数値)
    proc2 = run_sudo_cmd(["chmod", oct(mode)[2:], path])
    if proc2.returncode != 0:
        raise RuntimeError(
            f"sudo chmod failed: {proc2.stderr.decode(errors='ignore')}"
        )


def sudo_file_read(path: str) -> bytes:
    """
    sudo 経由でファイルの中身を読み取る。

    Returns
    -------
    bytes
        ファイル内容のバイト列。
    """
    proc = run_sudo_cmd(["cat", path])
    if proc.returncode != 0:
        raise RuntimeError(
            f"sudo read failed: {proc.stderr.decode(errors='ignore')}"
        )
    return proc.stdout


def sudo_path_exists(path: str) -> bool:
    """
    sudo 経由でパスの存在を確認する。

    `test -e path` の終了ステータスをチェックし、
    0 (存在する) なら True、0 以外 (存在しない等) なら False を返す。
    """
    proc = run_sudo_cmd(["test", "-e", path])
    return proc.returncode == 0


def sudo_mkdir_and_chmod(path: str, mode: int = 0o700) -> None:
    """
    sudo 経由でディレクトリを作成し、権限を変更する。

    - mkdir -p なので、既に存在していてもエラーにはならない。
    - その後 chmod で mode (例: 0o700) を強制的に設定する。
    """
    proc = run_sudo_cmd(["mkdir", "-p", path])
    if proc.returncode != 0:
        raise RuntimeError(
            f"sudo mkdir failed: {proc.stderr.decode(errors='ignore')}"
        )

    proc2 = run_sudo_cmd(["chmod", oct(mode)[2:], path])
    if proc2.returncode != 0:
        raise RuntimeError(
            f"sudo chmod failed: {proc2.stderr.decode(errors='ignore')}"
        )


# ======================================================================
# root(ローカル)で直接行うファイル/ディレクトリ操作
# ======================================================================

def ensure_dir_local(path: str) -> None:
    """
    ディレクトリをローカルで作成し、権限を 700 にする。

    - exist_ok=True なので既に存在する場合もエラーにしない。
    """
    os.makedirs(path, exist_ok=True)
    os.chmod(path, 0o700)


def write_file_local(path: str, data: bytes, mode: int = 0o600) -> None:
    """
    ローカルで原子的にファイルを上書きする。

    1. 同じディレクトリに一時ファイルを作成 (mkstemp)
    2. 一時ファイルにデータを書き込み、fsync() でディスクにフラッシュ
    3. 一時ファイルに chmod で mode を設定
    4. os.replace(一時ファイル, 本番ファイル)
       → OS レベルで「名前の付け替え」が行われるため、
         読む側から見ると「古いファイル → 一気に新しいファイル」に切り替わる

    これにより停電・例外等があっても、中途半端に書かれたファイルを
    本番として残さないようにできる。
    """
    dirpath = os.path.dirname(path)
    os.makedirs(dirpath, exist_ok=True)

    # umask を一時的に 077 にして、他ユーザに読み書きされないようにする
    old_umask = os.umask(0o077)
    try:
        import tempfile

        fd, tmppath = tempfile.mkstemp(prefix=".tmp.", dir=dirpath)
        try:
            with os.fdopen(fd, "wb") as f:
                f.write(data)
                f.flush()
                os.fsync(f.fileno())

            os.chmod(tmppath, mode)
            os.replace(tmppath, path)  # 原子的置換
        except Exception:
            # 失敗した場合は一時ファイルを掃除してから再度例外を投げる
            try:
                if os.path.exists(tmppath):
                    os.remove(tmppath)
            except Exception:
                pass
            raise
    finally:
        # umask を元に戻す
        os.umask(old_umask)


def read_file_local(path: str) -> bytes:
    """ローカルでファイルを開いて中身を読み取る。"""
    with open(path, "rb") as f:
        return f.read()


def delete_file_secure(path: str) -> bool:
    """
    ファイルをそれなりに安全に削除するユーティリティ。
    - 存在しなくても True を返す。
    - root の場合:os.remove で直接削除。
    - 非 root の場合:sudo rm -f で削除を試みる。
    ※専用ツールとかで復元できる消し方で、完全には消えない。
    """
    try:
        if is_root():
            if os.path.exists(path):
                os.remove(path)
            return True
        else:
            proc = run_sudo_cmd(["rm", "-f", path])
            if proc.returncode != 0:
                print(f"sudo rm に失敗: {proc.stderr.decode(errors='ignore')}")
                return False
            return True
    except Exception as e:
        print(f"ファイル削除中にエラー: {e}")
        return False


def delete_file_secure_hard(path: str) -> bool:
    """
    中身を上書きしてから削除する版。
    - ファイルが存在しなければ True を返す。
    - root の場合:
        * Python からファイルを開いて全サイズを 1〜2回上書き
        * fsync でディスクへのフラッシュを試みてから unlink
    - 非 root の場合:
        * まず sudo shred を試す
        * 失敗したら sudo rm -f にフォールバック

    ※ SSD / ジャーナリング FS / スナップショット / バックアップなどがある環境では
      「絶対に復元不能」とまでは言えない。Python からできる範囲でのハード削除。
    """
    try:
        if is_root():
            # root なら Python から直接上書き
            if not os.path.exists(path):
                return True

            try:
                st = os.stat(path)
                size = st.st_size
            except FileNotFoundError:
                return True

            if size > 0:
                # 読み書きバイナリで開く(バッファリングなし)
                with open(path, "r+b", buffering=0) as f:
                    # 1回目: 0x00 で埋める
                    patterns = [b"\x00"]
                    # ランダム bytes も足す
                    try:
                        patterns.append(os.urandom(1))
                    except Exception:
                        pass

                    for p in patterns:
                        f.seek(0)
                        block = p * 4096
                        remaining = size
                        while remaining > 0:
                            n = 4096 if remaining >= 4096 else remaining
                            f.write(block[:n])
                            remaining -= n
                        f.flush()
                        os.fsync(f.fileno())

            # 中身を上書きし終えたらファイルを削除
            os.remove(path)
            return True

        else:
            # 非 root の場合は sudo でハード削除を試みる
            # shred があればそれを使う(-u: unlink, -n 1: 1回上書き, -z: 最後にゼロで上書き)
            proc = run_sudo_cmd(["shred", "-u", "-n", "1", "-z", path])
            if proc.returncode == 0:
                return True

            # shred がない、または失敗した場合は rm -f にフォールバック
            proc2 = run_sudo_cmd(["rm", "-f", path])
            if proc2.returncode != 0:
                print(f"sudo rm (fallback) に失敗: {proc2.stderr.decode(errors='ignore')}")
                return False
            return True

    except Exception as e:
        print(f"delete_file_secure_hard 中にエラー: {e}")
        return False


# ======================================================================
# PASS_FILE の存在確認(root / sudo)
# ======================================================================

def has_passphrase_file_root() -> bool:
    """root で PASS_FILE の存在を確認する。"""
    return os.path.exists(PASS_FILE)


def has_passphrase_file_sudo() -> bool:
    """sudo 経由で PASS_FILE の存在を確認する。"""
    return sudo_path_exists(PASS_FILE)


# ======================================================================
# メニューから呼ばれる key 関数群
# ======================================================================

def key1() -> None:
    """
    [1] READ_FILE

    PASS_FILE の中身を表示する。
    - root の場合: read_file_local
    - 非 root の場合: sudo_file_read
    """
    print(f"[READ_FILE] {PASS_FILE}")
    try:
        if is_root():
            if not has_passphrase_file_root():
                print("ファイルが存在しません。")
                return
            data = read_file_local(PASS_FILE)
        else:
            if not has_passphrase_file_sudo():
                print("ファイルが存在しません。")
                return
            data = sudo_file_read(PASS_FILE)

        text = data.decode("utf-8", errors="replace")
        print("----- FILE CONTENT BEGIN -----")
        print(text)
        print("----- FILE CONTENT END -----")
    except Exception as e:
        print(f"読み取りに失敗しました: {e}")


def key2() -> None:
    """
    [2] WRITE_FILE

    PASS_FILE に内容を書き込む。
    - root の場合: ensure_dir_local + write_file_local
    - 非 root の場合: sudo_mkdir_and_chmod + sudo_file_write
    """
    print(f"[WRITE_FILE] {PASS_FILE}")
    try:
        import getpass

        text = getpass.getpass(
            "書き込む内容(パスフレーズなど)を入力してください: "
        )
        data = text.encode("utf-8")

        if is_root():
            ensure_dir_local(PASS_DIR)
            write_file_local(PASS_FILE, data, mode=0o600)
        else:
            sudo_mkdir_and_chmod(PASS_DIR, mode=0o700)
            sudo_file_write(PASS_FILE, data, mode=0o600)

        print("書き込みに成功しました。")
    except Exception as e:
        print(f"書き込みに失敗しました: {e}")


def key3() -> None:
    """
    [3] CREATE_FILE

    PASS_FILE が存在しない場合に「空ファイル」を作成する。
    既に存在する場合は何もしない。
    """
    print(f"[CREATE_FILE] {PASS_FILE}")
    try:
        if is_root():
            if has_passphrase_file_root():
                print("ファイルは既に存在します。何も行いません。")
                return
            ensure_dir_local(PASS_DIR)
            write_file_local(PASS_FILE, b"", mode=0o600)
        else:
            if has_passphrase_file_sudo():
                print("ファイルは既に存在します。何も行いません。")
                return
            sudo_mkdir_and_chmod(PASS_DIR, mode=0o700)
            sudo_file_write(PASS_FILE, b"", mode=0o600)

        print("空ファイルを作成しました。")
    except Exception as e:
        print(f"CREATE_FILE に失敗しました: {e}")


def key4() -> None:
    """
    [4] CREATE_DIR

    PASS_DIR を作成(既にある場合は権限を 700 に整える)。
    """
    print(f"[CREATE_DIR] {PASS_DIR}")
    try:
        if is_root():
            ensure_dir_local(PASS_DIR)
        else:
            sudo_mkdir_and_chmod(PASS_DIR, mode=0o700)

        print("ディレクトリを作成(または既存を 700 に調整)しました。")
    except Exception as e:
        print(f"CREATE_DIR に失敗しました: {e}")


def key5() -> None:
    """
    [5] DELETE_DIR

    PASS_DIR ごと中身を削除する。
    - root の場合: shutil.rmtree
    - 非 root の場合: sudo rm -rf
    """
    print(f"[DELETE_DIR] {PASS_DIR}")
    try:
        if is_root():
            if os.path.isdir(PASS_DIR):
                shutil.rmtree(PASS_DIR)
                print("ディレクトリを削除しました。")
            else:
                print("ディレクトリが存在しません。")
        else:
            proc = run_sudo_cmd(["rm", "-rf", PASS_DIR])
            if proc.returncode != 0:
                print(
                    f"sudo rm -rf に失敗: "
                    f"{proc.stderr.decode(errors='ignore')}"
                )
            else:
                print("ディレクトリを削除しました。(sudo)")
    except Exception as e:
        print(f"DELETE_DIR 中にエラー: {e}")


def key6() -> None:
    """
    [6] DELETE_FILE

    PASS_FILE だけを削除する(可能な範囲で中身を上書きしてから削除)。
    - 内部では delete_file_secure_hard() を利用。
    """
    print(f"[DELETE_FILE] {PASS_FILE}")
    ok = delete_file_secure_hard(PASS_FILE)
    if ok:
        print("PASS_FILE を削除しました。(中身も可能な限り上書き済み)")
    else:
        print("PASS_FILE の削除に失敗しました。")


# ======================================================================
# メインループ
# ======================================================================

def main() -> None:
    """メニューを表示して対話的に操作するメイン関数。"""
    print("=== TEST PASS FILE TOOL ===")
    print(f"PASS_DIR : {PASS_DIR}")
    print(f"PASS_FILE: {PASS_FILE}")
    print(f"実行ユーザ: {'root' if is_root() else 'non-root'}")
    print()

    while True:
        print("==== MENU ====")
        print("[1] READ_FILE   (PASS_FILE を読む)")
        print("[2] WRITE_FILE  (PASS_FILE に書く)")
        print("[3] CREATE_FILE (空の PASS_FILE を作る)")
        print("[4] CREATE_DIR  (PASS_DIR を作成)")
        print("[5] DELETE_DIR  (PASS_DIR を削除)")
        print("[6] DELETE_FILE (PASS_FILE を削除)")
        print("[7] END         (終了)")
        input_key = input("番号を入力してください >> ").strip()

        match input_key:
            case "1":
                key1()
            case "2":
                key2()
            case "3":
                key3()
            case "4":
                key4()
            case "5":
                key5()
            case "6":
                key6()
            case "7":
                print("正常終了。")
                sys.exit(0)
            case _:
                print("無効な値です。")


if __name__ == "__main__":
    main()

3.メモ

subprocess.run(args,input,stdout,stderr)…外部のコマンドを実行し、その結果をまとめて返してくれる関数。引数は以下の通り。※他にもあるけど最低限のものを記載しておきます

  • args…実行したいコマンドの文字列やリスト。文字列等の次にカンマ区切りで「shell=true」を書くとシェルを経由して実行するが、どのシェルかはわからず危険なため、使用しない
  • input…標準入力に渡したいデータ(bytes)。「b”文字列”」で指定する必要がある。これの説明は実例を見た方が分かりやすいと思う。
    inputを指定せずに「subprocess.run([“cat”,”./test.txt”],None,省略,省略)」を実行すると、「cat ./test.txt」となり、catはtest.txtをそのまま読み込んで内容をstdoutに入れるだけ。
    inputに「data=b”hello Python”」としたdataを指定して「subprocess.run([“cat”],data,省略,省略)」で実行すると「cat 」となり、catの中にdataが入力されることになる。cat等に何か入力したいときに使う。※catは引数無しで標準入力を読む
  • stdout…標準出力の保管先。「=PIPE」とすることでPythonへ渡している
  • stderr…標準エラー出力の保管先。「=PIPE」とすることでPythonへ渡している

subprocess.CompletedProcess…subprocess.run()の戻り値。以下が入っている

  • returncode…外部コマンドの終了ステータス(0→成功、1→失敗)。test等はboolが入る
  • stdout…外部コマンドを実行した標準出力が入っている
  • stderr…外部コマンドを実行した標準エラー出力が入っている

shlex.quote(文字列)…文字列を完全に “ただの文字列” に変換し、シェルに解釈されないようにする関数。これによりコマンドラインに渡す引数を1つにしている。

0o600…8進数の600。Linuxの権限は0~7で表すため、慣習として8進数とする。600は所有者のみ読み書きできるという意味

run_sudo_cmd([“bash”, “-c”, f”umask 077; cat > {safe_path}”],input_data=data,)…これの意味は、「sudo bash -c “umask 077; cat > /root/.test/pass.txt”」。「bash -c」は「umask」とcatのリダイレクトを使うために指定(Linuxのbuiltinはプロセス内でのみ有効)。
「umask 077」でグループとその他ユーザのアクセスを禁止。「cat > ファイル」で標準入力をファイルに出力。sudoでできる最もセキュアで簡素なコードがこれだと思う。

proc2 = run_sudo_cmd([“chmod”, oct(mode)[2:], path])…umaskをしても600権限にならないことがあるため、念のためchmodで権限を変更しておく。なお、「oct(mode)[2:]」は「0o600」の「0o」を読み飛ばして、chmodで扱える型にするためのもの。

os.makedirs(path, exist_ok=bool)…動作的にはmkdirと同じ。権限があるならこっちの方が完結で速い。「exist_ok=」をtrueにするとディレクトリが作成済みでもエラーにしない。

os.chmod(path,8進数3桁の権限 )…chmodと同じ。ファイル/ディレクトリのパーミッションを設定する。権限があればこっちの方が完結で速い。

os.path.dirname(path)…親ディレクトリの取り出し。

os.umask(8進数3桁の権限)…OSのumask自体を変更。現在の umask を返しつつ、指定した値に変更するため、これの返却値を変数に保管しておき、再度この関数で変数を読み込むことで、umaskを元に戻せる。

tempfile…安全な一時ファイルを作るための標準モジュール。セキュアな処理(パスワード・秘密鍵・暗号化データなど)を扱うプログラムでは必須

tempfile.mkstemp(prefix=拡張子, dir=path)…衝突しないユニークな一時ファイルをumaskに従って作る。返却値はファイルディスクリプタ=fd(ファイルの整数番号)と一時ファイルの絶対パスの2つ。fdはfdopen()等でファイルとして扱える。パスを使わないファイルアクセスで便利

os.fdopen(fd, モード)…指定したfdを指定モード(書き込みモードの”wb”等)で開く。必ず開いたら閉じる必要があるため、withで囲うことを推奨する

ファイルオブジェクト.write(書き込みたいデータ)…ファイルオブジェクトにデータを書き込む。正確にはPythonバッファに溜め込まれているだけのため、例外発生で実際のファイルに更新されなかったりする

ファイルオブジェクト.flush()…Python バッファを OS バッファに流す。OSに書き込み要求しただけで、まだ実際のファイルに更新されない

ファイルオブジェクト.fileno()…Pythonのファイルオブジェクトからfdを返す

ファイルオブジェクト.fsync(fd)…指定したfdのファイルにデータを完全にディスクへ強制書き込み。本文の書き方の話だが、os.fsync(fd)ではなくos.fsync(f.fileno())なのはwith内のライフサイクルに合わせるためでこちらの方が安全。os.fsync(fd)でも動くっちゃ動く

os.replace(ファイルパス1, ファイルパス2)…ファイルパス2をファイルパス1に強制的に置き換える。tmppath に完全なデータを仕上げてから、公式ファイルにパッと着替えさせている

os.stat(ファイル・ディレクトリパス)…ファイルやディレクトリのメタデータオブジェクトを取得。この中の「st_size」でファイルバイト数を取得している。これで上書きするバイト数を検出している。

shredコマンド…Linuxのrmではデータがディスクに残るが、これは復元困難な削除をする。

オプション意味
-u上書きのあと **削除(unlink)**する
-n N上書き回数を N 回にする
-z最後に 0 埋め(目立たなくする)
-v進捗表示

shutil.rmtree(ファイル・ディレクトリパス)…普通のrmdirは空ディレクトリしか消せないが、これは中が残っていようが、サブフォルダがあろうが全て再帰的に削除する。間違えて重要なフォルダを消したりすることもあるため、普段使いは厳禁。

タイトルとURLをコピーしました