一个使用FN Connect四级域名的MinIO做图床的尝试
承接上篇内容，我们已经通过 FN Connect + Docker 成功部署了 MinIO 本地对象存储，并实现了外网静态图片访问。但整套方案存在致命短板：仅能在 MinIO 后台手动上传图片，项目前端仅可通过 9000 接口展示图片，无法完成上传、删除、查询等管理操作。
本篇将深度剖析问题根源，提供自研 Flask 中转代理脚本 + Nginx 优化的完整解决方案，解决 MinIO SDK 与 FN Connect 网关鉴权冲突，实现图床完整增删改查（CRUD）能力。
温馨提示：本方案依旧为纯折腾整活方案，无特殊需求不必使用，生产环境优先使用阿里云OSS、腾讯云COS。
一、现有方案核心缺陷与问题根源
1.1 未优化方案的使用局限
- 只读不可写：外网可通过 FN Connect 四级域名直接访问、展示公开图片，但无法通过程序接口上传、删除图片
- 操作受限：所有图片管理操作只能手动登录 MinIO 9001 后台完成，无法对接前后端项目自动化调用
- SDK 完全失效：标准 MinIO Python/Java SDK 无法直接调用公网接口，请求直接被 FN Connect 网关拦截
1.2 底层冲突原理（核心）
FN Connect 公网访问网关自带Cookie 会话鉴权机制，所有外网请求必须携带合法的飞牛账号 Cookie，否则直接返回「暂无权限访问该服务」403 错误。
而官方 MinIO SDK 严格遵循 S3 标准签名协议，发起接口请求时，仅携带 S3 签名 Authorization 请求头，不支持自定义追加 Cookie。
强行修改 SDK 请求头追加 Cookie 会破坏 S3 签名校验规则，导致鉴权失败，因此直接改造 SDK 的方案完全不可行。
1.3 无效方案排除
- 关闭 FN Connect 鉴权：彻底**奔暴露 MinIO 接口，极易被恶意刷库、上传恶意文件，安全风险极高
- 更换第三方穿透：违背本方案「纯 FNOS 原生服务、零额外部署」的核心初衷，不适用于校园网、多层 NAT 无公网环境
二、最终优化架构设计
2.1 整体架构拓扑
客户端(前后端项目) → FN Connect网关 → Nginx(前置转发) → Flask中转脚本(NAS内网) → 本地MinIO服务
2.2 模块分工
- MinIO 原生服务(9000/9001)：仅负责静态图片公网只读展示，对外开放匿名下载权限
- Flask 中转代理脚本：部署于 NAS 内网，规避 FN 网关鉴权，承接所有图片上传、删除、查询、下载 CRUD 操作
- Nginx：统一请求入口、处理跨域、限流优化、拥有四级域名、隐藏后端脚本端口，提升访问稳定性
- 客户端请求：放弃原生 MinIO SDK，使用 Requests 携带 FN Connect 合法 Cookie 访问代理接口，顺利通过网关鉴权
三、完整部署实现
3.1 Flask 中转代理脚本（核心源码）
该脚本运行于 NAS 内网，直接调用本地 MinIO 服务，无需经过 FN Connect 穿透，彻底规避 Cookie 鉴权问题，支持上传、下载、删除、文件信息查询全功能。
from flask import Flask, request, jsonify
import base64
from io import BytesIO
from minio import Minio
from minio.error import S3Error
import logging
# 日志配置，方便排查
logging.basicConfig(level=logging.INFO)
app = Flask(__name__)
# 内网MinIO地址，必须使用内网IP/127.0.0.1，不走公网穿透
MINIO_ENDPOINT = "192.168.1.107:9000"
MINIO_SECURE = False
# 统一校验参数工具
def check_params(data: dict, required: list):
for k in required:
if k not in data or not str(data[k]).strip():
raise ValueError(f"参数 {k} 不能为空")
@app.route("/minio_api", methods=["POST"])
@app.route("/minio_api/", methods=["POST"])
def minio_api():
try:
payload = request.get_json()
if not payload:
return jsonify({"code": 400, "msg": "请求必须为JSON格式"})
data = payload.get("data", {})
# 必传基础参数校验
check_params(data, ["access_key", "secret_key", "bucket", "op"])
ak = data["access_key"]
sk = data["secret_key"]
bucket = data["bucket"]
op = data["op"]
# 初始化内网MinIO客户端
minio_client = Minio(MINIO_ENDPOINT, access_key=ak, secret_key=sk, secure=MINIO_SECURE)
# 桶不存在自动创建
if not minio_client.bucket_exists(bucket):
minio_client.make_bucket(bucket)
# 上传文件（base64传输，适配前后端项目）
if op == "upload":
check_params(data, ["object_name", "file_base64"])
obj_name = data["object_name"]
b64_str = data["file_base64"]
bin_data = base64.b64decode(b64_str)
stream = BytesIO(bin_data)
minio_client.put_object(bucket, obj_name, stream, length=len(bin_data))
app.logger.info(f"上传成功 bucket={bucket} obj={obj_name}")
return jsonify({"code": 200, "msg": "上传成功"})
# 下载文件（返回base64）
elif op == "download":
check_params(data, ["object_name"])
obj_name = data["object_name"]
resp = minio_client.get_object(bucket, obj_name)
raw = resp.read()
b64_out = base64.b64encode(raw).decode()
resp.close()
app.logger.info(f"下载成功 bucket={bucket} obj={obj_name} size={len(raw)}")
return jsonify({"code": 200, "data": {"file_base64": b64_out}, "msg": "ok"})
# 删除文件
elif op == "delete":
check_params(data, ["object_name"])
obj_name = data["object_name"]
minio_client.remove_object(bucket, obj_name)
app.logger.info(f"删除成功 bucket={bucket} obj={obj_name}")
return jsonify({"code": 200, "msg": "删除成功"})
# 查询文件元信息（大小、修改时间）
elif op == "stat":
check_params(data, ["object_name"])
obj_name = data["object_name"]
stat_info = minio_client.stat_object(bucket, obj_name)
res_data = {
"size": stat_info.size,
"last_modified": stat_info.last_modified.isoformat()
}
return jsonify({"code": 200, "data": res_data, "msg": "ok"})
# 不支持的操作类型
else:
return jsonify({"code": 400, "msg": f"不支持操作op={op}"})
# MinIO S3异常捕获
except S3Error as e:
if e.code == "NoSuchKey":
return jsonify({"code": 404, "err": "文件不存在", "msg": "操作失败"})
elif e.code == "NoSuchBucket":
return jsonify({"code": 404, "err": "桶不存在", "msg": "操作失败"})
return jsonify({"code": 500, "err": f"S3异常:{e.code}:{e.message}", "msg": "存储服务错误"})
# 参数校验异常
except ValueError as e:
return jsonify({"code": 400, "err": str(e), "msg": "参数错误"})
# 全局未知异常
except Exception as e:
app.logger.error(f"全局异常: {str(e)}", exc_info=True)
return jsonify({"code": 500, "err": str(e), "msg": "服务内部错误"})
if __name__ == "__main__":
# 0.0.0.0 允许内网Nginx跨服务访问
app.run(host="0.0.0.0", port=8080, debug=False)
3.2 环境依赖安装
SSH 连接 FNOS NAS，执行以下命令安装脚本所需依赖：
pip install flask minio requests
3.3 脚本后台常驻运行
# 后台运行脚本，日志输出到文件
nohup python3 minio_proxy.py > minio_proxy.log 2>&1
运行后脚本常驻 NAS 内网 8080 端口，等待 Nginx 转发请求。
3.4 Nginx 反向代理配置
通过 Nginx 统一入口转发请求，优化网络访问、解决跨域问题，配置如下：
server {
listen 8090;
server_name _;
# 统一接口路由，所有请求走 /minio_api
location /minio_api/ {
# 转发FN宿主机本机8080运行的Python中转脚本
proxy_pass http://host.docker.internal:8080/minio_api;
# 完整透传请求头
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# 超大二进制文件适配（Base64大包）
client_max_body_size 100G;
proxy_request_buffering off;
proxy_send_timeout 300s;
proxy_read_timeout 300s;
}
}
ta本身是docker，是这样部署的：
version: "3"
services:
minio-gateway-nginx:
image: nginx:alpine
container_name: minio-gateway-nginx
restart: always
# 宿主机8090 映射容器内8090
ports:
- "8090:8090"
volumes:
- ./conf:/etc/nginx/conf.d
# Linux FNOS必须加，容器内访问宿主机127.0.0.1:8080
extra_hosts:
- "host.docker.internal:host-gateway"
3.5 FN Connect 端口穿透配置
在 FNOS 后台 FN Connect 中新增端口转发规则：
- 内网目标地址：NAS 内网 IP
- 内网端口：8090
- 穿透类型：TCP+HTTP
配置完成后，即可通过 FN 四级域名公网访问代理接口。
四、客户端调用方案（核心解决403权限问题）
客户端放弃原生 minio SDK，使用 requests 携带 FN Connect 合法 Cookie 访问代理接口，完美通过网关鉴权。
4.1 Cookie 获取方式
- 浏览器登录 FNOS 飞牛桌面
- F12 打开开发者工具 → Application → Cookies → 对应 fnos.net 域名
- 复制完整会话 Cookie，加入请求请求头
- 我还写了一个自动化脚本来获取，用浏览器登录获取Cookie
"""
Playwright 自动登录 FNOS
两步：界面1输入 FN ID → 界面2输入账号密码 → 收 cookie
"""
import asyncio
import json
import os
COOKIE_FILE = os.path.join(os.path.dirname(__file__), "fnos_cookies.json")
URL = "https://FNID.fnos.net/login"
USER = ""
PASS = ""
def maybe_click(el):
"""尝试点击，可能是 el.click() 或 el.dispatchEvent(new MouseEvent('click'))"""
if el:
el.click()
async def main():
from playwright.async_api import async_playwright
async with async_playwright() as p:
browser = await p.chromium.launch(headless=False)
ctx = await browser.new_context()
page = await ctx.new_page()
# ---------- 第 0 步：打开登录页 ----------
print("[0] goto login...")
await page.goto(URL, wait_until="domcontentloaded", timeout=60000)
# ---------- 第 1 步：等待并填入 FN ID ----------
print("[1] wait for FN ID input...")
try:
await page.wait_for_selector("input", timeout=30000)
except:
print(" no input appeared, screenshot -> step_fail.png")
await page.screenshot(path="step_fail.png", full_page=True)
await asyncio.sleep(3)
inputs = await page.query_selector_all("input")
print(f" {len(inputs)} input(s) found")
for el in inputs:
t = (await el.get_attribute("type") or "")
p = (await el.get_attribute("placeholder") or "")
print(f" type={t} placeholder={p}")
# 填 FN ID (just in case)
fid_input = None
for el in inputs:
p = (await el.get_attribute("placeholder") or "")
n = (await el.get_attribute("name") or "")
if "FN" in p.upper() or "id" in n.lower() or "connect" in n.lower():
fid_input = el
break
if not fid_input and inputs:
fid_input = inputs[0]
if fid_input:
await fid_input.fill("chumc123")
print(" filled chumc123")
await fid_input.press("Enter")
await asyncio.sleep(5)
# ---------- 第 2 步：等待账号密码界面 ----------
print("[2] wait for username/password form...")
await asyncio.sleep(5)
inputs = await page.query_selector_all("input")
print(f" {len(inputs)} input(s) now")
user_el = pwd_el = None
for el in inputs:
t = (await el.get_attribute("type") or "")
p = (await el.get_attribute("placeholder") or "")
print(f" type={t} placeholder={p}")
if t in ("text", "email"):
user_el = el
elif t == "password":
pwd_el = el
elif "账号" in p or "手机" in p or "用户" in p:
user_el = el
elif "密码" in p:
pwd_el = el
if not user_el and len(inputs) >= 2:
user_el = inputs[0]
pwd_el = inputs[1]
if user_el and pwd_el:
await user_el.fill(USER)
await pwd_el.fill(PASS)
print(" filled chumc / password")
# 找登录按钮
btn = await page.query_selector("button")
if btn:
await btn.click()
print(" clicked login button")
else:
await pwd_el.press("Enter")
print(" pressed Enter")
else:
print(" could not find user/pass inputs")
# ---------- 第 3 步：给你 60 秒手动验证（2FA） ----------
print("[3] waiting 60s for you to complete verification...")
print(" (type 2FA code etc. in the browser window)")
await asyncio.sleep(60)
# ---------- 收 cookie ----------
cookies = await ctx.cookies()
print(f"[4] {len(cookies)} cookies total")
targets = {}
for c in cookies:
n = c["name"]
if n in ("entry-token", "fnos-token", "fnos-long-token", "mode", "language"):
targets[n] = c["value"]
print(f" {n} = {c['value'][:30]}...")
if targets:
with open(COOKIE_FILE, "w", encoding="utf-8") as f:
json.dump(targets, f, indent=2)
print(f"[OK] saved to {COOKIE_FILE}")
else:
print("[FAIL] no target cookies, saving all cookies to debug_cookies.json")
all_cookies = {c["name"]: c["value"] for c in cookies}
with open("debug_cookies.json", "w", encoding="utf-8") as f:
json.dump(all_cookies, f, indent=2)
print(" check debug_cookies.json")
# 不要让浏览器闪退
print("\nPress ENTER in this terminal to close browser...")
input()
await browser.close()
asyncio.run(main())
4.2 通用调用示例
"""
远程 MinIO 操作脚本
自动读取/刷新 FNOS Cookie → 调用 FNOS 上的 Flask API
"""
import os
import sys
import json
import requests
import subprocess
API_URL = "https://e****af-0.FNID.fnos.net/minio_api/"
AK = "L****T"
SK = "P****b"
BUCKET = "test01"
COOKIE_FILE = "fnos_cookies.json"
LOGIN_SCRIPT = "login_fnos.py"
def load_or_refresh_cookies():
"""读取 cookie 文件，不存在或过期则自动登录"""
if os.path.isfile(COOKIE_FILE):
with open(COOKIE_FILE, "r") as f:
cookies = json.load(f)
# 快速验证：用 cookie 试一次 API
if check_cookies(cookies):
return cookies
print("[Cookie] 已过期，重新登录...")
else:
print("[Cookie] 文件不存在，开始登录...")
return refresh_cookies()
def refresh_cookies():
"""调用 Playwright 登录获取新 cookie"""
script = os.path.join(os.path.dirname(__file__), LOGIN_SCRIPT)
if not os.path.isfile(script):
print(f"[错误] 找不到 {script}")
print("请先运行: pip install playwright && playwright install chromium")
sys.exit(1)
result = subprocess.run([sys.executable, script], capture_output=True, text=True)
print(result.stdout)
if result.returncode != 0:
print(f"[错误] 登录脚本失败:\n{result.stderr}")
sys.exit(1)
if not os.path.isfile(COOKIE_FILE):
print("[错误] 登录后未生成 cookie 文件")
sys.exit(1)
with open(COOKIE_FILE, "r") as f:
return json.load(f)
def check_cookies(cookies):
"""用 API 快速验证 cookie 是否有效"""
cookie_str = "; ".join(f"{k}={v}" for k, v in cookies.items())
try:
r = requests.post(API_URL, json={"data": {"op": "stat",
"object_name": "test.png", "access_key": AK, "secret_key": SK, "bucket": BUCKET}},
headers={"Cookie": cookie_str, "Content-Type": "application/json"}, timeout=10)
return r.status_code == 200
except:
return False
def call_api(op, **extra):
"""通用请求封装"""
cookies = load_or_refresh_cookies()
cookie_str = "; ".join(f"{k}={v}" for k, v in cookies.items())
payload = {"data": {"access_key": AK, "secret_key": SK, "bucket": BUCKET, "op": op, **extra}}
r = requests.post(API_URL, json=payload,
headers={"Cookie": cookie_str, "Content-Type": "application/json"},
timeout=300)
return r.json()
def stat(file_name):
res = call_api("stat", object_name=file_name)
if res.get("code") == 200:
d = res["data"]
print(f"[文件信息] {file_name} 大小:{d['size']/1024:.1f}KB 修改时间:{d['last_modified']}")
else:
print(f"[失败] {res}")
return res
def upload(local_path, remote_name):
import base64
with open(local_path, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
res = call_api("upload", object_name=remote_name, file_base64=b64)
print(f"[上传] {remote_name} 状态:{res.get('code')} {res.get('msg','')}")
return res
def download(remote_name, save_path):
import base64
res = call_api("download", object_name=remote_name)
if res.get("code") == 200:
data = base64.b64decode(res["data"]["file_base64"])
with open(save_path, "wb") as f:
f.write(data)
print(f"[下载] {remote_name} → {save_path} ({len(data)/1024:.1f}KB)")
else:
print(f"[失败] {res}")
return res
def delete(remote_name):
res = call_api("delete", object_name=remote_name)
print(f"[删除] {remote_name} 状态:{res.get('code')} {res.get('msg','')}")
return res
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python miniotest01.py <操作> [参数...]")
print(" stat <文件名>")
print(" upload <本地路径> <远程文件名>")
print(" download <远程文件名> <保存路径>")
print(" delete <远程文件名>")
sys.exit(1)
op = sys.argv[1]
if op == "stat":
stat(sys.argv[2])
elif op == "upload":
upload(sys.argv[2], sys.argv[3])
elif op == "download":
download(sys.argv[2], sys.argv[3])
elif op == "delete":
delete(sys.argv[2])
else:
print(f"不支持的操作: {op}")
五、方案整体优劣总结
5.1 核心优势
- 完美适配 FN Connect 鉴权机制，Cookie过期自动重新获取，彻底解决 SDK 403 无权限问题
- 实现图床完整 CRUD 能力，从「只能看」升级为「可完整管理」
- 纯内网调用 MinIO，安全可控，无需关闭网关鉴权
- 搭配 Nginx 优化，访问稳定、支持跨域、适配前后端项目
- 完全基于上篇架构迭代，无需新增穿透工具、无需改原有配置
5.2 现存局限
- FN Connect 免费版 Cookie 有时效性，需定期刷新更新
- 依赖 NAS 24 小时开机断 电，设备下线则图床失效
- 免费版带宽有限，不适合高并发、大文件场景
六、最终郑重提醒
本**方案（上下篇）仅为技术折腾、学习研究使用，流程繁琐、维护成本高。个人项目、生产环境、稳定需求场景，强烈优先使用阿里云 OSS、腾讯云 COS，免费额度足够日常使用，稳定省心。
如果想部署前后端+数据库+这个MinIO做二进制文件管理的话，将它们做成整个docker-compose就可以了，这样就少了大量的跨网传输，比现在单MinIO好太多，比如说类似以下docker-compose.yml
version: '3.8'
services:
# MySQL8.4 业务库 root/123456
mysql:
image: docker.1ms.run/mysql:8.4
container_name: hospital-mysql
restart: always
environment:
MYSQL_ROOT_PASSWORD: 123456
MYSQL_DATABASE: hospital
TZ: Asia/Shanghai
volumes:
- ./mysql/data:/var/lib/mysql
- ./mysql/init:/docker-entrypoint-initdb.d
- ./mysql/backup:/backup
ports:
- "3307:3306"
networks:
- hospital-network
command: --default-authentication-plugin=mysql_native_password --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
healthcheck:
test: ["CMD", "mysqladmin", "ping", "-uroot", "-p123456"]
interval: 5s
timeout: 3s
retries: 10
deploy:
resources:
limits:
cpus: '1.0'
memory: 1G
# Redis7 缓存
redis:
image: docker.1ms.run/redis:7-alpine
container_name: hospital-redis
restart: always
volumes:
- ./redis/data:/data
ports:
- "6380:6379"
networks:
- hospital-network
command: redis-server --appendonly yes --require redis123456
healthcheck:
test: ["CMD", "redis-cli", "-a", "redis123456", "ping"]
interval: 3s
timeout: 2s
retries: 8
deploy:
resources:
limits:
cpus: '0.5'
memory: 512M
# MinIO 文件存储
minio:
image: docker.1ms.run/minio/minio:latest
container_name: hospital-minio
restart: always
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin123
TZ: Asia/Shanghai
volumes:
- ./minio/data:/data
ports:
- "9002:9000"
- "9001:9001"
networks:
- hospital-network
command: server /data --console-address ":9001"
healthcheck:
test: ["CMD", "mc", "ready", "local"]
interval: 5s
timeout: 3s
retries: 5
deploy:
resources:
limits:
cpus: '0.5'
memory: 1G
# 容器1：纯JDK21 SpringBoot后端（无Python）
backend:
build: ./backend
container_name: hospital-backend
restart: on-failure
depends_on:
mysql:
condition: service_healthy
redis:
condition: service_healthy
minio:
condition: service_healthy
environment:
TZ: Asia/Shanghai
SPRING_PROFILES_ACTIVE: docker
volumes:
- ./backend/logs:/app/logs
ports:
- "8082:8080"
networks:
- hospital-network
healthcheck:
test: ["CMD", "wget", "--no-verbose", "--tries=1", "http://127.0.0.1:8080/actuator/health", "-O", "/dev/null"]
interval: 8s
timeout: 4s
retries: 6
deploy:
resources:
limits:
cpus: '1.5'
memory: 2G
# 容器2：纯Python3.10 定时任务/测试脚本容器（替代原来mysql-backup）
script-task:
build: ./task
container_name: hospital-script-task
restart: always
depends_on:
mysql:
condition: service_healthy
environment:
TZ: Asia/Shanghai
MYSQL_HOST: mysql
MYSQL_USER: root
MYSQL_PASSWORD: 123456
MYSQL_DB: hospital
volumes:
- ./task/scripts:/task/scripts
- ./task/logs:/task/logs
- ./mysql/backup:/backup
networks:
- hospital-network
deploy:
resources:
limits:
cpus: '0.3'
memory: 512M
# Nginx 前端反向代理
nginx:
image: docker.1ms.run/nginx:alpine
container_name: hospital-nginx
restart: always
ports:
- "81:80"
volumes:
- ./frontend/dist:/usr/share/nginx/html
- ./nginx/conf:/etc/nginx/conf.d
- ./nginx/logs:/var/log/nginx
depends_on:
backend:
condition: service_healthy
networks:
- hospital-network
healthcheck:
test: ["CMD", "nginx", "-t"]
interval: 10s
timeout: 3s
retries: 3
deploy:
resources:
limits:
cpus: '0.3'
memory: 256M
networks:
hospital-network:
driver: bridge
有缘人！你怎么知道我把字变白了？
本人fw大专生，现在大二哩。我觉得我太懒了，到现在还没有找到实习（bushi），我想找点有关Java前后端，大数据分析，Linux运维，安装售后工程师，也可以进入一些团队来接外包（我现在一个人，想有人能够吸纳我·呜呜呜）我自己大学期间就参加过大数据的技能大赛，拿到了江苏省省二等奖，我自己就研究研究关于Python，Java，C++，捣鼓一些奇奇怪怪的程序和项目，有研究MC的模组，平时打打游戏（很菜），MBTI是INFP，一只小蝴蝶，所以我还是比较懒，慢，不要太着急了哟，所以我对自己也很轻松，包活就行，自己攒攒钱升级我的服务器（对，能不能让我把我的dellr730xd搬到工作室里，帮我承担一下电费啊?阿里嘎多）我希望是能够找到我感兴趣的工作（对，兴趣比较重要，在包活的前提下，兴趣是第一位的）希望公司在长三角，出差可以接受全球（bushi，全国）可飞。
诶，什么时候可以让我不要被家里人催，干着自己爱干的事情啊？
