"""Native publication backends. Only Worker calls these methods. SFTP: verified immutable directories and an atomic POSIX symlink switch. S3: verified immutable object prefixes with a conditional current.json pointer. S3 requires a pointer-aware edge/gateway; it is not a naked S3 website switch. """ from __future__ import annotations import hashlib, io, json, os, posixpath, re, stat, uuid, zipfile from contextlib import contextmanager from pathlib import Path from urllib.parse import urlsplit from fastapi import HTTPException from . import secret_refs, safe_http from .deploy_agent import validate_archive SITE_RE=re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$') RELEASE_RE=re.compile(r'^[0-9a-f]{20}$') MAX_ARCHIVE=120*1024*1024 def _env(target,field): return os.getenv(f'SITEFORGE_DEPLOY_{target.upper()}_{field}','').strip() def backend(target): if target not in ('staging','production'):raise HTTPException(400,'未知发布环境') value=_env(target,'TYPE') or 'agent' if value not in ('agent','sftp','s3'):raise HTTPException(503,'发布目标类型无效') return value def _root(value): if not value.startswith('/') or value=='/' or '\\' in value or '..' in value.split('/'): raise HTTPException(503,'远程发布根目录必须为安全的绝对路径') normalized=posixpath.normpath(value) if normalized in ('/','/tmp') or len(normalized)<4:raise HTTPException(503,'远程发布根目录范围过大') return normalized def config(target, *, check_files=False): kind=backend(target) if kind=='agent':return {'type':'agent'} public=_env(target,'PUBLIC_URL') parsed=urlsplit(public) if parsed.scheme!='https' or not parsed.netloc or parsed.username or parsed.password: raise HTTPException(503,'原生 SFTP/S3 发布必须提前配置 HTTPS PUBLIC_URL') if kind=='sftp': host=_env(target,'HOST');user=_env(target,'USERNAME') root=_root(_env(target,'REMOTE_ROOT')) known=_env(target,'KNOWN_HOSTS_FILE');key=_env(target,'SSH_KEY_FILE') if not host or not user or not known or not key or '/' in host or ':' in host: raise HTTPException(503,'SFTP 缺少主机、用户、known_hosts 或私钥配置') try:port=int(_env(target,'PORT') or 22) except ValueError:raise HTTPException(503,'SFTP 端口无效') if not 1<=port<=65535:raise HTTPException(503,'SFTP 端口无效') if check_files and (not Path(known).is_file() or not Path(key).is_file()): raise HTTPException(503,'SFTP 私钥或 known_hosts 文件未挂载') return {'type':kind,'host':host,'user':user,'root':root,'known_hosts':known,'ssh_key':key,'port':port} bucket=_env(target,'BUCKET');region=_env(target,'REGION');prefix=_env(target,'PREFIX') or 'siteforge' if not bucket or not region or not re.fullmatch(r'[a-zA-Z0-9][a-zA-Z0-9._-]{1,62}',bucket): raise HTTPException(503,'S3 存储桶或区域配置不完整') if prefix.startswith('/') or '..' in prefix.split('/') or not re.fullmatch(r'[a-zA-Z0-9/_-]+',prefix): raise HTTPException(503,'S3 对象前缀不安全') endpoint=_env(target,'ENDPOINT_URL') if endpoint: parsed=urlsplit(endpoint) if parsed.scheme!='https' or not parsed.netloc or parsed.username or parsed.password or parsed.path.strip('/'): raise HTTPException(503,'S3 兼容服务端点必须为 HTTPS 根地址') if _env(target,'POINTER_AWARE')!='1': raise HTTPException(503,'S3 必须配置可解析 current.json 的发布网关(POINTER_AWARE=1)') return {'type':kind,'bucket':bucket,'region':region,'prefix':prefix.strip('/'), 'endpoint':endpoint} def _archive(raw:bytes): if len(raw)>MAX_ARCHIVE:raise ValueError('超出单次发布大小上限') validate_archive(raw) output={} with zipfile.ZipFile(io.BytesIO(raw)) as z: for member in z.infolist(): if member.is_dir():continue path=member.filename # validate_archive already checked all paths, duplicates, symlinks and size bounds. output[path]=z.read(member) hashes={k:hashlib.sha256(v).hexdigest() for k,v in output.items()} digest=hashlib.sha256(raw).hexdigest() return digest[:20],digest,output,hashes def _manifest(digest,hashes): return json.dumps({'sha256':digest,'files':hashes},sort_keys=True,separators=(',',':')).encode() def _probe(target,site,body): public_url=_env(target,'PUBLIC_URL') if not public_url: raise HTTPException(503,'正式发布必须配置站点 PUBLIC_URL 以执行上线探测') parsed=urlsplit(public_url) if parsed.scheme!='https' or not parsed.netloc or parsed.username or parsed.password: raise HTTPException(503,'PUBLIC_URL 必须为 HTTPS 网站地址') url=public_url.rstrip('/')+'/' try: response=safe_http.fetch(url,timeout=25,max_bytes=8_000_000) if response.status_code!=200 or hashlib.sha256(response.content).digest()!=hashlib.sha256(body).digest(): raise HTTPException(502,'上线首页内容与构建版本不一致') except safe_http.OutboundError as exc: raise HTTPException(502,'上线地址不可访问,需检查发布和 DNS/CDN 状态') from exc return True # ---------------- SFTP ---------------- @contextmanager def _sftp_connection(cfg): """Strict host-key verification; private key mounted into Worker, never transmitted to API.""" try:import paramiko except ImportError as exc:raise HTTPException(503,'缺少可选的 Paramiko SFTP 依赖') from exc client=paramiko.SSHClient() client.load_host_keys(cfg['known_hosts']) client.set_missing_host_key_policy(paramiko.RejectPolicy()) try: client.connect(cfg['host'],port=cfg['port'],username=cfg['user'], key_filename=cfg['ssh_key'],look_for_keys=False,allow_agent=False, timeout=12,auth_timeout=12,banner_timeout=12) conn=client.open_sftp() try:yield conn finally:conn.close() finally:client.close() def _sftp_type(sftp,path): try:return sftp.lstat(path).st_mode except OSError:return None def _mkdirs(sftp,path): cur='' for part in path.split('/'): if not part:continue cur+='/'+part mode=_sftp_type(sftp,cur) if mode is None: sftp.mkdir(cur) mode=_sftp_type(sftp,cur) if not mode or not stat.S_ISDIR(mode):raise ValueError('远端目录包含符号链接或非目录') def _read(sftp,path,limit=4*1024*1024): with sftp.file(path,'rb') as f:raw=f.read(limit+1) if len(raw)>limit:raise ValueError('远程校验文件过大') return raw def _write(sftp,path,data): with sftp.file(path,'wb') as f:f.write(data);f.flush() def _current_sftp(sftp,root): pointer=root+'/current' mode=_sftp_type(sftp,pointer) if mode is None:return None if not stat.S_ISLNK(mode):raise ValueError('current 不是受管符号链接') link=sftp.readlink(pointer) if not re.fullmatch(r'releases/[0-9a-f]{20}',link):raise ValueError('current 指向不安全') return link.rsplit('/',1)[1] def _verify_sftp(sftp,root,release): folder=root+'/releases/'+release mode=_sftp_type(sftp,folder) if mode is None or not stat.S_ISDIR(mode):raise ValueError('远程版本目录缺失或不安全') raw=_read(sftp,folder+'/.siteforge-manifest.json') obj=json.loads(raw) checks=obj.get('files') if not isinstance(checks,dict) or 'index.html' not in checks:raise ValueError('远端文件清单无效') for name,expected in checks.items(): if name.startswith('/') or '..' in name.split('/') or not re.fullmatch('[a-f0-9]{64}',str(expected)): raise ValueError('远端文件清单包含非法路径或哈希') mode=_sftp_type(sftp,folder+'/'+name) if mode is None or not stat.S_ISREG(mode):raise ValueError('远端发布资源缺失或不是普通文件') with sftp.file(folder+'/'+name,'rb') as f: dig=hashlib.sha256() while True: chunk=f.read(256*1024) if not chunk:break dig.update(chunk) if dig.hexdigest()!=expected:raise ValueError('远端文件哈希不匹配') return obj def _activate_sftp(sftp,root,release): previous=_current_sftp(sftp,root) candidate=root+'/current-'+release+'.next' if _sftp_type(sftp,candidate) is not None:raise ValueError('发现冲突的发布指针临时文件') sftp.symlink('releases/'+release,candidate) try: # OpenSSH posix-rename is atomic and replaces current. Fail closed if unsupported. sftp.posix_rename(candidate,root+'/current') finally: if _sftp_type(sftp,candidate) is not None:sftp.remove(candidate) if _current_sftp(sftp,root)!=release:raise ValueError('发布指针切换未确认') return previous def deploy_sftp(target,site,raw,cfg=None): cfg=cfg or config(target,check_files=True) release,digest,files,hashes=_archive(raw) root=cfg['root'].rstrip('/')+'/'+site with _sftp_connection(cfg) as sftp: _mkdirs(sftp,root+'/releases') folder=root+'/releases/'+release if _sftp_type(sftp,folder) is None: tmp=root+'/releases/.staging-'+release+'-'+uuid.uuid4().hex[:12] _mkdirs(sftp,tmp) try: for name,body in files.items(): parent=posixpath.dirname(name) if parent:_mkdirs(sftp,tmp+'/'+parent) _write(sftp,tmp+'/'+name,body) _write(sftp,tmp+'/.siteforge-manifest.json',_manifest(digest,hashes)) # Immutable version path is not overwritten on retry. sftp.rename(tmp,folder) except Exception: # No traffic change on failure. Interrupted transfers leave an # isolated temporary directory; a retry uses a fresh name. raise meta=_verify_sftp(sftp,root,release) if meta.get('sha256')!=digest:raise ValueError('现有版本摘要与当前构建不一致') previous=_current_sftp(sftp,root) if previous!=release:_activate_sftp(sftp,root,release) try:_probe(target,site,files['index.html']) except Exception: if previous and previous!=release: with _sftp_connection(cfg) as sftp: _verify_sftp(sftp,root,previous) _activate_sftp(sftp,root,previous) raise return {'published':True,'release':release,'previous':previous,'files':len(files),'sha256':digest,'target':target,'backend':'sftp'} def rollback_sftp(target,site,release,cfg=None,expected_home=None): if not RELEASE_RE.fullmatch(release):raise ValueError('无效的版本号') cfg=cfg or config(target,check_files=True) root=cfg['root'].rstrip('/')+'/'+site with _sftp_connection(cfg) as sftp: _verify_sftp(sftp,root,release) previous=_current_sftp(sftp,root) if previous!=release:_activate_sftp(sftp,root,release) content=_read(sftp,root+'/releases/'+release+'/index.html',8*1024*1024) try:_probe(target,site,expected_home if expected_home is not None else content) except Exception: if previous and previous!=release: with _sftp_connection(cfg) as sftp:_activate_sftp(sftp,root,previous) raise return {'rolled_back':True,'release':release,'backend':'sftp','target':target} # ---------------- S3 / compatible object store ---------------- def _s3_client(cfg): try:import boto3 except ImportError as exc:raise HTTPException(503,'缺少可选的 boto3 S3 依赖') from exc return boto3.client('s3',region_name=cfg['region'],endpoint_url=cfg['endpoint'] or None) def _s3_key(cfg,site,path):return cfg['prefix']+'/sites/'+site+'/'+path def _s3_bytes(s3,bucket,key,max_bytes=4*1024*1024): response=s3.get_object(Bucket=bucket,Key=key) data=response['Body'].read(max_bytes+1) if len(data)>max_bytes:raise ValueError('S3 读取超过限制') return data def _s3_current(s3,cfg,site): key=_s3_key(cfg,site,'current.json') try: response=s3.get_object(Bucket=cfg['bucket'],Key=key) except Exception as exc: code=getattr(exc,'response',{}).get('Error',{}).get('Code','') if str(code) in ('404','NoSuchKey','NotFound'):return None,None raise data=response['Body'].read(4096) obj=json.loads(data) if not RELEASE_RE.fullmatch(obj.get('release','')):raise ValueError('当前 S3 发布指针无效') return obj['release'],response.get('ETag') def _s3_verify(s3,cfg,site,release): path=_s3_key(cfg,site,'releases/'+release+'/') obj=json.loads(_s3_bytes(s3,cfg['bucket'],path+'.siteforge-manifest.json')) checks=obj['files'] if 'index.html' not in checks:raise ValueError('S3 版本没有入口页') for name,digest in checks.items(): if name.startswith('/') or '..' in name.split('/') or not re.fullmatch('[a-f0-9]{64}',str(digest)): raise ValueError('S3 文件清单路径或哈希非法') response=s3.get_object(Bucket=cfg['bucket'],Key=path+name) hasher=hashlib.sha256() while True: chunk=response['Body'].read(256*1024) if not chunk:break hasher.update(chunk) if hasher.hexdigest()!=digest:raise ValueError('S3 对象校验失败') return obj def _s3_point(s3,cfg,site,release,previous,etag): key=_s3_key(cfg,site,'current.json') body=json.dumps({'release':release,'previous':previous},sort_keys=True).encode() args={'Bucket':cfg['bucket'],'Key':key,'Body':body,'ContentType':'application/json','CacheControl':'no-store'} if etag:args['IfMatch']=etag else:args['IfNoneMatch']='*' s3.put_object(**args) current,_=_s3_current(s3,cfg,site) if current!=release:raise ValueError('S3 发布指针未确认') def _s3_immutable_put(s3,bucket,key,body,**extra): """Create-only object write. A retry can reuse only byte-identical keys.""" try: s3.put_object(Bucket=bucket,Key=key,Body=body,IfNoneMatch='*',**extra) return except Exception as exc: code=str(getattr(exc,'response',{}).get('Error',{}).get('Code','')) if code not in ('PreconditionFailed','412','ConditionalRequestConflict','409'): raise existing=_s3_bytes(s3,bucket,key,max_bytes=max(8_000_000,len(body))) if hashlib.sha256(existing).digest()!=hashlib.sha256(body).digest(): raise ValueError('现有 S3 对象与当前发布文件不一致,禁止覆盖') def deploy_s3(target,site,raw,cfg=None): cfg=cfg or config(target) release,digest,files,hashes=_archive(raw) s3=_s3_client(cfg);bucket=cfg['bucket'] base=_s3_key(cfg,site,'releases/'+release+'/') manifest_key=base+'.siteforge-manifest.json' try: exists=_s3_bytes(s3,bucket,manifest_key) except Exception as exc: code=getattr(exc,'response',{}).get('Error',{}).get('Code','') if str(code) not in ('404','NoSuchKey','NotFound'):raise exists=None if exists is None: for name,body in files.items(): import mimetypes ct=mimetypes.guess_type(name)[0] or 'application/octet-stream' cache='no-cache' if name.endswith('.html') else 'public, max-age=31536000, immutable' _s3_immutable_put(s3,bucket,base+name,body,ContentType=ct,CacheControl=cache) _s3_immutable_put(s3,bucket,manifest_key,_manifest(digest,hashes), ContentType='application/json',CacheControl='no-store') manifest=_s3_verify(s3,cfg,site,release) if manifest.get('sha256')!=digest:raise ValueError('S3 版本摘要冲突') previous,etag=_s3_current(s3,cfg,site) if previous!=release:_s3_point(s3,cfg,site,release,previous,etag) try:_probe(target,site,files['index.html']) except Exception: if previous and previous!=release: current,next_etag=_s3_current(s3,cfg,site) if current==release:_s3_point(s3,cfg,site,previous,release,next_etag) raise return {'published':True,'release':release,'previous':previous,'files':len(files),'sha256':digest,'target':target,'backend':'s3'} def rollback_s3(target,site,release,cfg=None,expected_home=None): if not RELEASE_RE.fullmatch(release):raise ValueError('无效的版本号') cfg=cfg or config(target) s3=_s3_client(cfg) _s3_verify(s3,cfg,site,release) previous,etag=_s3_current(s3,cfg,site) if previous!=release:_s3_point(s3,cfg,site,release,previous,etag) body=_s3_bytes(s3,cfg['bucket'],_s3_key(cfg,site,'releases/'+release+'/index.html'),8*1024*1024) try:_probe(target,site,expected_home if expected_home is not None else body) except Exception: if previous and previous!=release: current,new_etag=_s3_current(s3,cfg,site) if current==release:_s3_point(s3,cfg,site,previous,release,new_etag) raise return {'rolled_back':True,'release':release,'target':target,'backend':'s3'} def deploy(target,site,archive): if not SITE_RE.fullmatch(site):raise ValueError('无效的网站标识') cfg=config(target,check_files=True) raw=Path(archive).read_bytes() if cfg['type']=='sftp':return deploy_sftp(target,site,raw,cfg) if cfg['type']=='s3':return deploy_s3(target,site,raw,cfg) raise ValueError('不是原生发布目标') def rollback(target,site,release): if not SITE_RE.fullmatch(site):raise ValueError('无效的网站标识') cfg=config(target,check_files=True) if cfg['type']=='sftp':return rollback_sftp(target,site,release,cfg) if cfg['type']=='s3':return rollback_s3(target,site,release,cfg) raise ValueError('不是原生发布目标')