#!/usr/bin/env python3 """ conoha-ftp.py — ConoHa WING Skill 用 FTPS 操作ツール (v1.0.3 / ftplib版) v2 からの変更点(レビュー指摘対応): [安全] 設定ファイルは「このスクリプトと同じフォルダ」を既定にし、実行場所(CWD)依存で 認証情報入り conf が OneDrive/Git 配下に作られる事故を防ぐ。 [安全] macOS Keychain 保存で失敗した場合でも、例外文字列にパスワードが混じって 表示されないよう必ずマスクする(argv 経由の一時露出も最小化)。 [安全] リモートパスに改行・制御文字を許さない(FTP コマンドインジェクション/事故防止)。 [堅牢] 存在判定が確実にできない場合は「不明」とし、新規と決めつけて無承認で上書きしない。 [堅牢] 想定外エラーは生のトレースバックを出さず、利用者にやさしい一文へ変換(--debug で詳細)。 [堅牢] 一時ファイル名に PID を付けて実ファイルとの衝突を避け、失敗時は必ず後始末する。 [機能] フォルダごと公開する publish コマンド(1接続で再帰アップロード、削除はしない)。 [機能] 公開後に「見られるアドレス(URL)」を表示(conf の domain を使用)。 [機能] download はローカルの同名ファイルを無断上書きしない(--force で明示上書き)。 [互換] 日本語ファイル名のため FTP の文字コードを UTF-8 に固定(Python 3.8 対策)。 維持する安全策: - 証明書検証: ssl.create_default_context() を FTP_TLS に渡す(証明書+ホスト名検証) - データチャネル暗号化: 接続後に必ず prot_p() - 平文フォールバック禁止: 素の FTP は使わず FTP_TLS のみ + login 前に auth() - TLS セッション再利用: データ接続で制御接続の TLS セッションを再利用 - 封じ込め: '..'・絶対パス・docroot 外への書き込みを拒否 - 操作範囲: 読取/追加/上書きのみ。利用者向けの削除・同期コマンドは持たせない - サーバー管理ファイル(.htaccess / .user.ini 等)は一切アップロード/上書きしない - 上書きは2段階: upload --plan(提案) → 承認 → upload --execute --approved - アトミックアップロード: 一時ファイル名で put → 完了後にリネーム 設定ファイル例 (conoha-ftp.conf): [conoha] host = xxxxx.conohawing.com user = ftpアカウント名 password = ******** # 初回操作時に暗号化保存へ自動移行 docroot = # 接続許可ディレクトリ制限時は空 domain = example.com # 公開URL表示用(任意) 依存: Python 3.8+ のみ(外部ツール不要) """ import argparse import base64 import configparser import ftplib import getpass import os import posixpath import shutil import ssl import subprocess import sys from datetime import datetime, timezone # 設定ファイルはこのスクリプトと同じフォルダを既定にする(実行場所に依存しない)。 # → スキルフォルダ(~/.claude/skills/...) 内に固定され、CWD が OneDrive/Git でも巻き込まれない。 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) DEFAULT_CONF = os.path.join(SCRIPT_DIR, "conoha-ftp.conf") # ---------------------------------------------------------------------------- # TLS セッション再利用を強制する FTP_TLS サブクラス # ---------------------------------------------------------------------------- # 多くの FTPS サーバーは「データ接続のTLSセッションは制御接続のものを再利用せよ」と # 要求する(セッションスティーリング対策)。素の ftplib はデータ接続で新規セッションを # 張るため弾かれることがある。制御接続の SSL セッションをデータ接続に引き継ぐ。 class ReusedSSLFTP(ftplib.FTP_TLS): def ntransfercmd(self, cmd, rest=None): conn, size = ftplib.FTP.ntransfercmd(self, cmd, rest) if self._prot_p: # prot_p() 済みのときだけ TLS でラップ session = self.sock.session # 制御接続のセッションを再利用 conn = self.context.wrap_socket( conn, server_hostname=self.host, session=session ) return conn, size # ---------------------------------------------------------------------------- # 認証情報保護: アクセス権を本人のみに制限 / 同期フォルダ検出 # ---------------------------------------------------------------------------- def secure_conf_permissions(conf_path: str, verbose: bool = False) -> bool: """設定ファイルを本人のみアクセス可に制限する(冪等)。 posix: chmod 600 / Windows: icacls で継承を切り本人にのみフルアクセス付与。 認証情報のみが対象でパスワードは扱わないため、外部コマンド呼び出しでも安全。""" try: if os.name == "posix": os.chmod(conf_path, 0o600) if verbose: print("アクセス権を本人のみ(600)に制限しました。") return True if os.name == "nt": user = os.environ.get("USERNAME") or os.environ.get("USER") or "" if not user: if verbose: print("[注意] ユーザー名が取得できず、権限制限をスキップしました。", file=sys.stderr) return False # フェイルセーフ2段階: ①先に本人へフルアクセスを付与(本人を締め出さない)。 # 失敗したら継承を外さず中止する(継承だけ外れて本人すら読めない事故を防ぐ)。 g = subprocess.run( ["icacls", conf_path, "/grant:r", f"{user}:F"], capture_output=True, text=True, errors="replace", ) if g.returncode != 0: if verbose: print(f"[注意] 権限付与に失敗したため制限をスキップ: " f"{g.stderr.strip() or g.stdout.strip()}", file=sys.stderr) return False # ②本人許可が確実に入った後で、継承(他ユーザー分)を除去する。 r = subprocess.run( ["icacls", conf_path, "/inheritance:r"], capture_output=True, text=True, errors="replace", ) if r.returncode == 0: if verbose: print(f"アクセス権を本人({user})のみに制限しました。") return True if verbose: print(f"[注意] 継承の除去に失敗しました(本人のアクセスは確保済み): " f"{r.stderr.strip() or r.stdout.strip()}", file=sys.stderr) return False except Exception as e: if verbose: print(f"[注意] 権限設定中に問題: {e}", file=sys.stderr) return False def _looks_synced(conf_path: str) -> bool: """OneDrive/Dropbox 等の同期フォルダ配下かを簡易判定する。""" p = os.path.abspath(conf_path).replace("\\", "/").lower() return any(k in p for k in ("onedrive", "dropbox", "google drive", "googledrive", "icloud")) def _redact(text: str, secret: str) -> str: """文字列中のパスワードを伏字に置換する(ログ・例外メッセージ用)。""" if secret and text: return text.replace(secret, "********") return text # ---------------------------------------------------------------------------- # 認証情報の暗号化保存: OS の標準機構に委譲(pip 依存なし) # Windows = DPAPI(ctypes) / macOS = Keychain(security) / Linux = secret-tool # いずれも使えない場合は平文 conf + 権限制限にフォールバック(壊さない)。 # ---------------------------------------------------------------------------- _CRED_SERVICE = "conoha-wing-skill" def _cred_account(host: str, user: str) -> str: return f"{user}@{host}" def _secret_backend() -> str: """利用可能な暗号化保存バックエンド名。無ければ ''(=平文フォールバック)。""" if os.name == "nt": return "dpapi" if sys.platform == "darwin" and shutil.which("security"): return "keychain" if sys.platform.startswith("linux") and shutil.which("secret-tool"): return "secret-tool" return "" def _dpapi(protect: bool, data: bytes) -> bytes: """Windows DPAPI(ctypes)。本人＋同一PCでのみ復号可能な暗号化/復号。""" import ctypes from ctypes import wintypes class BLOB(ctypes.Structure): _fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_char))] buf = ctypes.create_string_buffer(data, len(data)) blob_in = BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char))) blob_out = BLOB() fn = (ctypes.windll.crypt32.CryptProtectData if protect else ctypes.windll.crypt32.CryptUnprotectData) # 第6引数 0x1 = CRYPTPROTECT_UI_FORBIDDEN(UIを出さない) if not fn(ctypes.byref(blob_in), None, None, None, None, 0x1, ctypes.byref(blob_out)): raise OSError("DPAPI 呼び出しに失敗しました") out = ctypes.string_at(blob_out.pbData, blob_out.cbData) ctypes.windll.kernel32.LocalFree(blob_out.pbData) return out def store_password(host: str, user: str, password: str) -> tuple: """パスワードを安全に保存し、conf に書くべき (キー, 値) を返す。 失敗時は ('password', 平文) を返す(フォールバック=壊さない)。 例外メッセージにパスワードが混じり得るため、必ずマスクしてから表示する。""" backend = _secret_backend() try: if backend == "dpapi": enc = base64.b64encode(_dpapi(True, password.encode("utf-8"))).decode("ascii") return ("password_enc", enc) if backend == "keychain": # 注意: security コマンドはパスワードを stdin で受け取れないため -w で渡す。 # ごく短時間 argv に載るが、失敗時の表示は必ずマスクする(下記 except)。 subprocess.run( ["security", "add-generic-password", "-U", "-s", _CRED_SERVICE, "-a", _cred_account(host, user), "-w", password], check=True, capture_output=True, text=True, errors="replace", ) return ("password_store", "keychain") if backend == "secret-tool": subprocess.run( ["secret-tool", "store", "--label", "ConoHa WING Skill", "service", _CRED_SERVICE, "account", _cred_account(host, user)], input=password, check=True, capture_output=True, text=True, errors="replace", ) return ("password_store", "secret-tool") except Exception as e: print(f"[注意] 暗号化保存に失敗したため平文保存にフォールバックします: " f"{_redact(str(e), password)}", file=sys.stderr) return ("password", password) def load_password(sec, host: str, user: str) -> str: """conf セクションから平文パスワードを取り出す(暗号化保存に対応)。""" if sec.get("password_enc"): try: return _dpapi(False, base64.b64decode(sec.get("password_enc"))).decode("utf-8") except Exception as e: die(f"保存されたパスワードの復号に失敗しました: {e}\n" f"別のユーザー/PC にコピーした場合は復号できません。`setup` で再設定してください。") store = (sec.get("password_store") or "").strip() if store == "keychain": r = subprocess.run( ["security", "find-generic-password", "-s", _CRED_SERVICE, "-a", _cred_account(host, user), "-w"], capture_output=True, text=True, errors="replace", ) if r.returncode == 0: return r.stdout.strip() die("Keychain からパスワードを取得できませんでした。`setup` で再設定してください。") if store == "secret-tool": r = subprocess.run( ["secret-tool", "lookup", "service", _CRED_SERVICE, "account", _cred_account(host, user)], capture_output=True, text=True, errors="replace", ) if r.returncode == 0 and r.stdout: return r.stdout.strip() die("secret-tool からパスワードを取得できませんでした。`setup` で再設定してください。") return (sec.get("password") or "").strip() # 平文(レガシー/フォールバック) def _migrate_plaintext_password(conf_path, cp, host, user, password) -> None: """conf に平文パスワードが残っている場合、暗号化保存へ自動移行する。""" if not _secret_backend(): return # 暗号化が使えない環境: 平文のまま(権限制限で保護) key, val = store_password(host, user, password) if key == "password": return # 保存に失敗 → 平文のまま継続 sec = cp["conoha"] sec.pop("password", None) sec[key] = val with open(conf_path, "w", encoding="utf-8") as f: cp.write(f) secure_conf_permissions(conf_path) print("パスワードを暗号化保存に移行しました(平文はファイルから削除済み)。", file=sys.stderr) # ---------------------------------------------------------------------------- # 設定読み込み + 検証 # ---------------------------------------------------------------------------- def load_config(conf_path: str) -> dict: if not os.path.exists(conf_path): die(f"設定ファイルが見つかりません: {conf_path}\n" f"`setup` コマンドで作成できます: python3 conoha-ftp.py setup") # 認証情報保護: アクセス権を本人のみに制限する(冪等。posix=600 / Windows=icacls)。 secure_conf_permissions(conf_path) if _looks_synced(conf_path): print( "[注意] 設定ファイルが同期フォルダ(OneDrive/Dropbox 等)にあるようです。" "認証情報が同期・流出する恐れがあるため、同期対象外の場所へ移動してください。", file=sys.stderr, ) cp = configparser.ConfigParser() cp.read(conf_path, encoding="utf-8") if "conoha" not in cp: die("設定ファイルに [conoha] セクションがありません。") sec = cp["conoha"] # docroot は任意。空 = 接続許可ディレクトリのルート直下で運用する。 # ConoHa WING で FTPアカウントを「接続許可ディレクトリ=public_html/<ドメイン>」に # 制限すると、ログイン起点がその公開フォルダ内に固定(chroot)される。その場合 # docroot を更に付けるとパスが二重になるため、docroot は空にする(=封じ込めは # サーバー側の接続許可ディレクトリが担う)。制限していない場合のみ docroot に # public_html/<ドメイン> 等を指定する。 docroot = sec.get("docroot", "").strip().rstrip("/") if docroot in (".", "/"): docroot = "" if docroot.startswith("./"): docroot = docroot[2:] host = sec.get("host", "").strip() user = sec.get("user", "").strip() domain = sec.get("domain", "").strip() if not host: die("設定ファイルに host が設定されていません。") if not user: die("設定ファイルに user が設定されていません。") password = load_password(sec, host, user) if not password: die("設定ファイルにパスワードが設定されていません。`setup` で作成してください。") cfg = {"host": host, "user": user, "password": password, "docroot": docroot, "domain": domain} # docroot を指定する場合のみ妥当性チェック(絶対パス・.. は不可)。 if cfg["docroot"] and ( cfg["docroot"].startswith("/") or cfg["docroot"].split("/")[0] == ".." ): die(f"docroot の指定が不正です: '{cfg['docroot']}'(絶対パスや .. は使えません)") # 平文パスワードが残っていれば、暗号化保存へ自動移行する(可能な環境のみ)。 if sec.get("password") and not sec.get("password_enc") and not sec.get("password_store"): _migrate_plaintext_password(conf_path, cp, host, user, password) return cfg # ---------------------------------------------------------------------------- # パス封じ込め: docroot 外・'..'・絶対パス・制御文字を拒否 # ---------------------------------------------------------------------------- def safe_remote_path(docroot: str, rel: str) -> str: # 改行・制御文字は FTP コマンドへの混入(インジェクション)や事故の元。明確に拒否。 if any(ord(c) < 0x20 or c == "\x7f" for c in rel): die("ファイル名に使用できない文字(改行・制御文字)が含まれています。") if rel.startswith("/"): die(f"絶対パスは指定できません: {rel}") if docroot: joined = posixpath.normpath(posixpath.join(docroot, rel)) if joined != docroot and not joined.startswith(docroot + "/"): die(f"書き込み先が許可範囲(docroot)外です: {rel}") else: # docroot 空 = ルート直下(接続許可ディレクトリ)。'..' での上位脱出のみ拒否。 joined = posixpath.normpath(rel) if joined == ".." or joined.startswith("../"): die(f"書き込み先が許可範囲(ルート)外です: {rel}") return joined # ---------------------------------------------------------------------------- # FTPS 接続(安全策をすべて明示) # ---------------------------------------------------------------------------- def connect(cfg: dict) -> ReusedSSLFTP: context = ssl.create_default_context() # 既定で証明書・ホスト名を検証 try: ftps = ReusedSSLFTP(context=context, timeout=30) ftps.encoding = "utf-8" # 日本語ファイル名対策(Python 3.8 は既定 latin-1) ftps.connect(cfg["host"], 21) # 明示的TLS(explicit)はポート21 ftps.auth() # 制御チャネルを TLS 化(平文ログイン防止) ftps.login(cfg["user"], cfg["password"]) ftps.prot_p() # データチャネルも暗号化(必須) ftps.set_pasv(True) # パッシブモード return ftps except ssl.SSLCertVerificationError as e: die(f"サーバー証明書の検証に失敗しました: {e}\n" f"host の指定が正しいか確認してください。") except ftplib.error_perm as e: die(f"ログインに失敗しました(認証エラー): {_redact(str(e), cfg['password'])}\n" f"user / password を確認してください。") except OSError as e: die(f"サーバーに接続できませんでした: {e}\n" f"host の指定とネットワークを確認してください。") def remote_exists(ftps: ReusedSSLFTP, remote: str) -> dict: """リモートファイルの状態を判定して返す。 返り値は {"status": "present"|"absent"|"unknown", "size":..., "mtime":...}。 重要: 判定が確実にできないときは "unknown" を返す。呼び出し側はこれを 「既存かもしれない」とみなし、利用者の承認なしに上書きしないこと (権限・サーバー仕様・一時障害を「不在」と誤認して無断上書きする事故を防ぐ)。 """ # --- 存在 + 更新日時: MDTM(ファイルにのみ成功。ディレクトリには失敗) --- mtime = None try: resp = ftps.sendcmd(f"MDTM {remote}") # 例: "213 20260610143000" if resp[:3] == "213": ts = resp[4:].strip() try: mtime = datetime.strptime(ts[:14], "%Y%m%d%H%M%S").replace( tzinfo=timezone.utc ).strftime("%Y-%m-%d %H:%M (UTC)") except ValueError: mtime = None return {"status": "present", "size": _safe_size(ftps, remote), "mtime": mtime} except ftplib.error_perm: pass # MDTM 非対応/不在の可能性。NLST で最終確認する。 # --- フォールバック: 親ディレクトリ一覧に名前があるか --- parent = posixpath.dirname(remote) or "." name = posixpath.basename(remote) try: names = ftps.nlst(parent) if any(posixpath.basename(n) == name for n in names): return {"status": "present", "size": _safe_size(ftps, remote), "mtime": None} # 一覧の取得に成功し、その中に無い → 確実に「不在」。 return {"status": "absent", "size": None, "mtime": None} except ftplib.error_perm: # 一覧自体が取れない。親が新規サブディレクトリなら「不在(新規)」とみなしてよいが、 # ルート直下で一覧が取れないのは異常なので「不明」にして安全側に倒す。 if parent in ("", "."): return {"status": "unknown", "size": None, "mtime": None} return {"status": "absent", "size": None, "mtime": None} def _safe_size(ftps: ReusedSSLFTP, remote: str): """サイズ取得(バイナリモードへ切替。失敗しても判定は壊さない)。""" try: ftps.voidcmd("TYPE I") return ftps.size(remote) except Exception: return None def ensure_remote_dirs(ftps: ReusedSSLFTP, remote: str): """remote(ファイルパス)の親ディレクトリを順に作成(無ければ)。""" parent = posixpath.dirname(remote) if not parent: return parts = parent.split("/") cur = "" for p in parts: cur = p if not cur else f"{cur}/{p}" try: ftps.mkd(cur) except ftplib.error_perm: pass def _silent_delete(ftps: ReusedSSLFTP, remote: str): """一時ファイルの後始末(失敗は無視)。利用者向けの delete 機能とは別物。""" try: ftps.delete(remote) except Exception: pass def _public_url(cfg: dict, rel: str): """conf の domain と相対パスから公開URLを組み立てる(domain 未設定なら None)。""" domain = cfg.get("domain") if not domain: return None rel = rel.lstrip("./") # index.html はディレクトリ表示になるので末尾を省く if rel == "index.html": return f"https://{domain}/" if rel.endswith("/index.html"): return f"https://{domain}/{rel[:-len('index.html')]}" return f"https://{domain}/{rel}" # ---------------------------------------------------------------------------- # アップロード本体(単一ファイル)。upload / publish で共用。 # ---------------------------------------------------------------------------- def _put_file(ftps: ReusedSSLFTP, local: str, remote: str): """1ファイルをアトミックにアップロードする。失敗時は一時ファイルを残さない。""" ensure_remote_dirs(ftps, remote) tag = os.getpid() tmp = f"{remote}.{tag}.uploading.tmp" ftps.voidcmd("TYPE I") # バイナリ転送(ASCII だと画像等が壊れる) try: with open(local, "rb") as f: ftps.storbinary(f"STOR {tmp}", f) except Exception: _silent_delete(ftps, tmp) # 中断時のゴミを残さない raise try: ftps.rename(tmp, remote) except ftplib.error_perm: # rename が既存へ直接上書きできないサーバー向け: 既存を退避してから入替。 backup = f"{remote}.{tag}.bak.tmp" ftps.rename(remote, backup) # ① 既存を退避 try: ftps.rename(tmp, remote) # ② 新規を本名へ except ftplib.error_perm as e: ftps.rename(backup, remote) # 失敗時は原状回復(元ファイルを戻す) _silent_delete(ftps, tmp) die(f"上書きに失敗しました: {e}\n" f"元のファイルは保持されています(変更はありません)。") _silent_delete(ftps, backup) # 退避した旧ファイルのみ後始末 # ---------------------------------------------------------------------------- # Tier 0: 一覧 # ---------------------------------------------------------------------------- def cmd_list(cfg: dict, args): remote = safe_remote_path(cfg["docroot"], args.path or "") ftps = connect(cfg) try: lines = [] ftps.dir(remote, lines.append) print("\n".join(lines) if lines else "(ファイルなし)") finally: ftps.quit() # ---------------------------------------------------------------------------- # Tier 0: ダウンロード(ローカルの無断上書きはしない) # ---------------------------------------------------------------------------- def cmd_download(cfg: dict, args): remote = safe_remote_path(cfg["docroot"], args.remote) if os.path.exists(args.local) and not args.force: die(f"保存先に同名ファイルがあります: {args.local}\n" f"上書きして良ければ --force を付けて再実行してください。") ftps = connect(cfg) try: ftps.voidcmd("TYPE I") part = args.local + ".part" # いったん別名で受けてから差し替え(中断時に既存を壊さない) with open(part, "wb") as f: ftps.retrbinary(f"RETR {remote}", f.write) os.replace(part, args.local) print(f"ダウンロード完了: {remote} -> {args.local}") finally: ftps.quit() # ---------------------------------------------------------------------------- # Tier 1/2: アップロード(--plan / --execute の2段階) # ---------------------------------------------------------------------------- def cmd_upload(cfg: dict, args): if not os.path.exists(args.local): die(f"ローカルファイルが見つかりません: {args.local}") if os.path.isdir(args.local): die(f"フォルダが指定されました。フォルダごと公開するには publish を使ってください: " f"python3 conoha-ftp.py publish {args.local} --plan") if _is_protected(args.remote): die(".htaccess などサーバーが管理するファイルは、このスキルでは扱いません。\n" "(上書きするとリダイレクト・高速化・PHP設定などが壊れる恐れがあるため)\n" "編集が必要な場合は、ConoHa WING コントロールパネルの「ファイルマネージャー」か、\n" "FTPソフト(FileZilla / WinSCP / Cyberduck など)で手動で操作してください。") remote = safe_remote_path(cfg["docroot"], args.remote) local_size = os.path.getsize(args.local) local_mtime = datetime.fromtimestamp( os.path.getmtime(args.local), tz=timezone.utc ).strftime("%Y-%m-%d %H:%M (UTC)") ftps = connect(cfg) try: info = remote_exists(ftps, remote) status = info["status"] needs_approval = status in ("present", "unknown") if args.plan: print("=== アップロード提案 ===") print(f"対象: {remote}") if status == "present": size_str = f"{info['size']:,} bytes" if info["size"] is not None else "サイズ不明" mtime_str = info["mtime"] or "更新日時不明" print(" 区分: 【上書き】(既存ファイルあり)") print(f" 既存: {size_str} / {mtime_str}") print(f" 新規: {local_size:,} bytes / {local_mtime}(今回アップロード)") print("\n※ 既存ファイルを上書きします。実行してよいか利用者の承認を得てください。") print(" 承認後: --execute --approved を付けて再実行してください。") elif status == "unknown": print(" 区分: 【確認できませんでした】(既存の可能性あり)") print(f" 新規: {local_size:,} bytes / {local_mtime}") print("\n※ サーバー上に同名ファイルがあるか確認できませんでした。" "安全のため上書き扱いとします。") print(" 実行する場合: 利用者の承認を得て --execute --approved を付けてください。") else: print(" 区分: 新規(上書きなし)") print(f" 新規: {local_size:,} bytes / {local_mtime}") print("\n新規追加のため、--execute で実行できます。") return if needs_approval and not args.approved: die("上書き(または存在不明)対象です。--plan で提案を提示し利用者の承認を得てから、" "--execute --approved を付けて実行してください。") _put_file(ftps, args.local, remote) verb = "新規アップロード" if status == "absent" else "上書き" print(f"{verb}完了: {args.local} -> {remote}") url = _public_url(cfg, args.remote) if url: print(f"公開アドレス: {url}") finally: ftps.quit() # ---------------------------------------------------------------------------- # Tier 1/2: フォルダごと公開(1接続で再帰アップロード。削除はしない) # ---------------------------------------------------------------------------- _SKIP_NAMES = {"conoha-ftp.conf", ".DS_Store", "Thumbs.db", "desktop.ini"} # サーバー(ConoHa)が自動生成・管理することがあるファイル。上書きするとリダイレクト・高速化・ # PHP設定・Basic認証などの設定が壊れる恐れがあるため、このスキルでは一切アップロード・上書き # しない(delete と同じく「手段を持たせない」)。 _PROTECTED_NAMES = {".htaccess", ".htpasswd", ".user.ini"} def _is_protected(name: str) -> bool: """サーバー管理ファイル(.htaccess 等)かどうか。basename で判定(大小無視)。""" return posixpath.basename(name.replace("\\", "/")).lower() in _PROTECTED_NAMES def _collect_files(localdir: str, remote_base: str): """localdir 配下のファイルを列挙する。 返り値: (items, skipped_protected) items = [(ローカルパス, リモート相対パス), ...] skipped_protected = サーバー管理のため除外した相対パス名のリスト(透明性のため報告する) junk(_SKIP_NAMES / *.tmp)は従来どおり無言でスキップする。""" items = [] skipped_protected = [] for root, dirs, names in os.walk(localdir): dirs[:] = [d for d in dirs if not d.startswith(".")] # .git / .well-known 等は除外 for n in sorted(names): rel = os.path.relpath(os.path.join(root, n), localdir).replace(os.sep, "/") if _is_protected(n): skipped_protected.append(rel) continue if n in _SKIP_NAMES or n.endswith(".tmp"): continue full = os.path.join(root, n) remote_rel = posixpath.join(remote_base, rel) if remote_base else rel items.append((full, remote_rel)) return items, skipped_protected def cmd_publish(cfg: dict, args): localdir = args.localdir if not os.path.isdir(localdir): die(f"フォルダが見つかりません: {localdir}") remote_base = safe_remote_path(cfg["docroot"], args.remote or "") # remote_base は docroot を含む正規化パス。各ファイルの相対は args.remote 起点で組む。 base_rel = (args.remote or "").strip("/") items, skipped_protected = _collect_files(localdir, base_rel) if not items: if skipped_protected: die(f"アップロード対象のファイルがありません: {localdir}\n" f"(サーバー管理ファイルのみ検出: {', '.join(skipped_protected)} は公開対象外です)") die(f"アップロード対象のファイルがありません: {localdir}") ftps = connect(cfg) try: plan = [] # (local, remote_full, rel, status) for local, rel in items: remote_full = safe_remote_path(cfg["docroot"], rel) status = remote_exists(ftps, remote_full)["status"] plan.append((local, remote_full, rel, status)) new_items = [p for p in plan if p[3] == "absent"] over_items = [p for p in plan if p[3] in ("present", "unknown")] if args.plan: print("=== フォルダ公開の提案 ===") print(f"対象フォルダ: {localdir} (ファイル数: {len(plan)})") print(f" 新規: {len(new_items)} 件 / 入れ替え(上書き・要承認): {len(over_items)} 件\n") if new_items: print("[新規]") for _l, _r, rel, _s in new_items: print(f" + {rel}") if over_items: print("\n[入れ替え(上書き)= 既存または確認できないファイル]") for _l, _r, rel, s in over_items: mark = "?" if s == "unknown" else "*" print(f" {mark} {rel}") print("\n※ 入れ替えがあります。利用者の承認を得てから --execute --approved を付けてください。") else: print("\nすべて新規です。--execute で公開できます。") if skipped_protected: print("\n[除外(サーバー管理のため公開しません)]") for rel in skipped_protected: print(f" - {rel}") return if over_items and not args.approved: die(f"入れ替え(上書き)対象が {len(over_items)} 件あります。--plan で内容を提示し" f"利用者の承認を得てから、--execute --approved を付けて実行してください。") done = 0 for local, remote_full, rel, status in plan: _put_file(ftps, local, remote_full) done += 1 print(f" ({done}/{len(plan)}) {rel}") print(f"\n公開完了: {done} 件のファイルをアップロードしました。") if skipped_protected: print(f"サーバー管理のため除外したファイル: {', '.join(skipped_protected)}") top_url = _public_url(cfg, base_rel + "/index.html" if base_rel else "index.html") if top_url: print(f"公開アドレス: {top_url}") finally: ftps.quit() # ---------------------------------------------------------------------------- # setup: 設定ファイルを対話生成 + 接続テスト # ---------------------------------------------------------------------------- def cmd_setup(cfg_unused, args): conf_path = args.conf if _looks_synced(conf_path): print("[注意] 設定ファイルの保存先が同期フォルダ(OneDrive/Dropbox 等)のようです。" "認証情報の流出を避けるため、同期対象外の場所での利用を推奨します。", file=sys.stderr) if os.path.exists(conf_path): ans = input(f"{conf_path} は既にあります。上書きしますか? [y/N]: ").strip().lower() if ans != "y": print("中止しました。") return print("ConoHa WING の接続情報を入力してください。") # プロンプトはコンパネ実画面の表記に合わせる(FTPサーバー / ユーザー名 …)。 host = input(" FTPサーバー (例: www1001.example-server.com): ").strip() user = input(" ユーザー名 (@以降も含め全体・例: test@example.com): ").strip() password = getpass.getpass(" パスワード (入力中は画面に表示されません): ") # ローカル入力のみ domain = input(" ドメイン (ユーザー名の@より後ろ・URL表示用/任意・例: example.com): ").strip() print() print("FTPアカウントの『接続許可ディレクトリ』を対象ドメインに制限していますか?") print("(推奨。制限すると ConoHa はそのドメインの公開フォルダ内に接続を固定します)") restricted = input(" 制限している? [Y/n]: ").strip().lower() if restricted in ("", "y", "yes"): # 制限あり = 接続起点が公開フォルダ。docroot は空(ルート直下)にする。 docroot = "" print(" -> docroot は空(ルート直下)に設定。封じ込めは接続許可ディレクトリが担います。") else: dom2 = domain or input(" 対象ドメイン名(例: example.com): ").strip() docroot = f"public_html/{dom2}" if not domain: domain = dom2 print(f" -> docroot = {docroot}") section = {"host": host, "user": user, "docroot": docroot} if domain: section["domain"] = domain key, val = store_password(host, user, password) section[key] = val cp = configparser.ConfigParser() cp["conoha"] = section with open(conf_path, "w", encoding="utf-8") as f: cp.write(f) secure_conf_permissions(conf_path, verbose=True) if key == "password_enc": print("パスワードは暗号化して保存しました(Windows DPAPI)。") elif key == "password_store": print(f"パスワードは OS の資格情報ストア({val})に保存しました。") else: print("[注意] 暗号化保存が使えない環境のため平文で保存します(権限制限で保護)。") print(f"\n{conf_path} を作成しました(docroot = {docroot or '(空=ルート直下)'})。") print("接続テストを実行します...") cfg = load_config(conf_path) ftps = connect(cfg) try: ftps.dir(cfg["docroot"] or ".", lambda _l: None) print("接続成功。公開先ディレクトリにアクセスできました。セットアップ完了です。") finally: ftps.quit() def die(msg: str): print(f"[エラー] {msg}", file=sys.stderr) sys.exit(1) def build_parser(): p = argparse.ArgumentParser( description="ConoHa WING Skill 用 FTPS 操作ツール(v1.0.3/ftplib)。" "読取・追加・上書きのみ。削除・同期は非対応。外部ツール不要。" ) p.add_argument("--conf", default=DEFAULT_CONF, help=f"設定ファイル(既定: {DEFAULT_CONF})") p.add_argument("--debug", action="store_true", help="想定外エラー時に詳しい技術情報(トレースバック)を表示する") sub = p.add_subparsers(dest="command", required=True) # --debug はサブコマンドの「後ろ」に付けても効くよう、共通の親パーサで各サブにも付与する。 # default=SUPPRESS にして、未指定時にメイン側の --debug 値を上書きしないようにする # (これで `... upload x y --debug` も `... --debug upload x y` も両方効く)。 common = argparse.ArgumentParser(add_help=False) common.add_argument("--debug", action="store_true", default=argparse.SUPPRESS, help="想定外エラー時に詳しい技術情報(トレースバック)を表示する") sp = sub.add_parser("setup", parents=[common], help="設定ファイルを対話作成し接続テストする") sp.set_defaults(func=cmd_setup) sp = sub.add_parser("list", parents=[common], help="[Tier0] docroot 配下の一覧") sp.add_argument("path", nargs="?", default="", help="docroot からの相対パス(省略可)") sp.set_defaults(func=cmd_list) sp = sub.add_parser("download", parents=[common], help="[Tier0] ファイルをダウンロード") sp.add_argument("remote", help="docroot からの相対パス") sp.add_argument("local", help="保存先ローカルパス") sp.add_argument("--force", action="store_true", help="ローカルに同名があっても上書きする") sp.set_defaults(func=cmd_download) sp = sub.add_parser("upload", parents=[common], help="[Tier1/2] 1ファイルをアップロード(新規/上書き)") sp.add_argument("local", help="アップロードするローカルファイル") sp.add_argument("remote", help="docroot からの相対パス(アップロード先)") g = sp.add_mutually_exclusive_group(required=True) g.add_argument("--plan", action="store_true", help="提案のみ表示(実行しない)") g.add_argument("--execute", dest="execute", action="store_true", help="実アップロード") sp.add_argument("--approved", action="store_true", help="上書き時の承認済みフラグ(--execute と併用)") sp.set_defaults(func=cmd_upload, plan=False) sp = sub.add_parser("publish", parents=[common], help="[Tier1/2] フォルダごと公開(1接続で再帰アップロード)") sp.add_argument("localdir", help="公開するローカルフォルダ") sp.add_argument("remote", nargs="?", default="", help="docroot からの相対サブパス(省略可。通常は省略)") g = sp.add_mutually_exclusive_group(required=True) g.add_argument("--plan", action="store_true", help="提案のみ表示(実行しない)") g.add_argument("--execute", dest="execute", action="store_true", help="実アップロード") sp.add_argument("--approved", action="store_true", help="入れ替え(上書き)を含む場合の承認済みフラグ") sp.set_defaults(func=cmd_publish, plan=False) return p def main(): parser = build_parser() args = parser.parse_args() try: if args.command == "setup": args.func(None, args) return cfg = load_config(args.conf) args.func(cfg, args) except SystemExit: raise # die() の終了はそのまま通す except KeyboardInterrupt: die("中断しました。") except Exception: if getattr(args, "debug", False): raise # --debug 時は詳細(トレースバック)を表示 die("うまくいきませんでした。もう一度お試しください。\n" "繰り返し失敗する場合は、接続情報(サーバー名・アカウント・パスワード)や、" "ネットワーク、FTPアカウントの接続許可ディレクトリ設定をご確認ください。\n" "(詳しい技術情報を見るには、コマンドの後ろに --debug を付けて再実行してください)") if __name__ == "__main__": main()