"""FIELD-AB-01: editable site profiles, separate secrets and explicit activation.""" import copy import hashlib import ipaddress import json import os from pathlib import Path import pwd import re import tempfile from urllib.parse import urlsplit SETUP = Path('/workspace/contextra/setup') PROFILES = SETUP / 'profiles' STATE = Path('/var/lib/contextra-maintenance/field') RUNTIME = Path('/workspace/contextra/runtime') def account(): return pwd.getpwnam('si') def safe_path(path): path = Path(path) if not path.is_absolute() or any(p.is_symlink() for p in (path, *path.parents)): raise ValueError('절대 경로만 허용하며 symlink는 사용하지 않습니다.') return path def read_json(path, limit=131072): path = safe_path(path) if not path.is_file() or path.stat().st_size > limit: raise ValueError('설정 파일이 없거나 너무 큽니다: ' + str(path)) return json.loads(path.read_text()) def atomic(path, data, *, user=False, mode=0o600): path = safe_path(path) path.parent.mkdir(parents=True, exist_ok=True, mode=0o700 if not user else 0o750) raw = data if isinstance(data, bytes) else (json.dumps(data, ensure_ascii=False, indent=2) + '\n').encode() fd, name = tempfile.mkstemp(prefix='.field-next-', dir=path.parent) try: with os.fdopen(fd, 'wb') as stream: os.fchmod(stream.fileno(), mode) if user and os.geteuid() == 0: u = account(); os.fchown(stream.fileno(), u.pw_uid, u.pw_gid) stream.write(raw); stream.flush(); os.fsync(stream.fileno()) os.replace(name, path) fd = os.open(path.parent, os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) finally: Path(name).unlink(missing_ok=True) def identity(value): if not isinstance(value, str) or not re.fullmatch(r'[a-z][a-z0-9_-]{0,39}', value): raise ValueError('이름은 소문자로 시작하는 영문/숫자/_/- 40자 이내입니다.') return value def port(value): if type(value) is not int or not 1 <= value <= 65535: raise ValueError('포트는 1..65535 정수입니다.') return value def ipv4(value): address = ipaddress.IPv4Address(value) if address.is_unspecified or address.is_multicast or address.is_loopback: raise ValueError('실제 unicast IPv4가 필요합니다.') return str(address) def url(value, *, origin=False): if not isinstance(value, str) or any(c.isspace() for c in value): raise ValueError('URL 형식 오류') parsed = urlsplit(value) if parsed.scheme not in (('https',) if origin else ('https', 'http')) or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment: raise ValueError('인증정보 없는 URL을 사용하세요. Manager는 HTTPS만 지원합니다.') if origin and parsed.path not in ('', '/'): raise ValueError('Manager 주소에는 API 경로를 넣지 않습니다.') if parsed.port is not None: port(parsed.port) return value.rstrip('/') if origin else value def keys(value, required, optional=()): if not isinstance(value, dict) or not set(required) <= value.keys() or set(value) - set(required) - set(optional): raise ValueError('설정 항목 오류: ' + ', '.join(required)) def validate(value): p = copy.deepcopy(value) keys(p, ('version','id','label','network','control_port','redis_port','redis_allow','media','webs','selected_web','artifact_lan_cidr')) if type(p['version']) is not int or p['version'] != 1: raise ValueError('지원하지 않는 프로필 version') identity(p['id']) if not isinstance(p['label'], str) or not p['label'] or len(p['label']) > 100: raise ValueError('환경 이름이 필요합니다.') n = p['network'] keys(n, ('interface','mode','address','gateway','dns','ssh_port')) if not isinstance(n['interface'], str) or not re.fullmatch(r'[a-zA-Z0-9_.:-]{1,32}', n['interface']): raise ValueError('NIC 이름 오류') if n['mode'] not in ('dhcp','static'): raise ValueError('mode=dhcp 또는 static') if n['address']: address = ipaddress.IPv4Interface(n['address']); ipv4(str(address.ip)) if address.ip in (address.network.network_address,address.network.broadcast_address) and address.network.prefixlen < 31: raise ValueError('호스트 주소가 아닙니다.') elif n['mode'] == 'static': raise ValueError('고정주소/prefix가 필요합니다.') if n['gateway']: ipv4(n['gateway']) if not isinstance(n['dns'], list) or len(n['dns']) > 4: raise ValueError('DNS는 최대4개') for a in n['dns']: ipv4(a) port(n['ssh_port']) for key in ('control_port','redis_port'): port(p[key]) if p[key] < 1024: raise ValueError('Contextra 외부 포트는 1024 이상입니다.') if len({p['control_port'],p['redis_port'],n['ssh_port']}) != 3: raise ValueError('포트가 겹칩니다.') for key in ('redis_allow','media'): if not isinstance(p[key], list) or len(p[key]) > 32: raise ValueError(key + ' 최대32개') for a in p['redis_allow']: ipv4(a) for m in p['media']: keys(m, ('label','host','port')); ipv4(m['host']); port(m['port']) if not isinstance(m['label'], str) or len(m['label']) > 100: raise ValueError('미디어 이름 오류') if not isinstance(p['webs'], list) or not 1 <= len(p['webs']) <= 16: raise ValueError('Web 후보는 1..16개') ids = set() for w in p['webs']: keys(w, ('id','label','host','manager','event_url','server_id'), ('connection_version',)) if 'connection_version' in w: from connection_versions import valid valid(w['connection_version']) identity(w['id']); ipv4(w['host']) if w['id'] in ids: raise ValueError('Web ID 중복') ids.add(w['id']); url(w['manager'], origin=True); url(w['event_url']) if urlsplit(w['event_url']).hostname != urlsplit(w['manager']).hostname or urlsplit(w['event_url']).path != '/api/v1/events': raise ValueError('Event는 선택 Manager의 /api/v1/events 계약입니다.') if not isinstance(w['server_id'], str) or not w['server_id'].strip() or len(w['server_id']) > 256: raise ValueError('AI server ID가 필요합니다.') if not isinstance(w['label'], str) or len(w['label']) > 100: raise ValueError('Web 이름 오류') if p['selected_web'] not in ids: raise ValueError('selected_web가 후보 목록에 없습니다.') cidr = ipaddress.IPv4Network(p['artifact_lan_cidr'], strict=False) if cidr.prefixlen < 8: raise ValueError('미디어 허용망이 너무 넓습니다.') if ipaddress.IPv4Address(web(p)['host']) not in cidr: raise ValueError('선택 Web이 미디어 허용망에 없습니다.') if web(p)['host'] not in p['redis_allow']: raise ValueError('선택 Web을 Redis 허용 주소에 포함하세요.') return p def web(p): return next(w for w in p['webs'] if w['id'] == p['selected_web']) def path(name): return PROFILES / (identity(name) + '.json') def load(name=None): if name is None: selected = read_json(STATE / 'selected.json') name = selected['profile'] p = validate(read_json(path(name))) if p['id'] != name: raise ValueError('파일명과 프로필 id가 다릅니다.') return p def save(p): p = validate(p) atomic(path(p['id']), p, user=True) return path(p['id']) def secret_path(p): return PROFILES / (p['id'] + '.private') / (web(p)['id'] + '.credentials.json') def read_credentials(path): from auth_control import login_id path=safe_path(path);info=path.stat() if info.st_mode & 0o077 or (info.st_uid!=0 and info.st_uid!=account().pw_uid): raise ValueError('계정 파일은 root 또는 si 소유600이어야 합니다.') value=read_json(path,16384) keys(value,('loginId','password'));login_id(value['loginId']) password=value['password'] if not isinstance(password,str) or not password or len(password.encode())>8192 or any(c in password for c in ('\x00','\r','\n')): raise ValueError('계정 파일 비밀번호 형식 오류') return value def ca_path(p): return PROFILES / (p['id'] + '.private') / (web(p)['id'] + '.ca.crt') def stamp(p): return hashlib.sha256(json.dumps(validate(p),sort_keys=True,ensure_ascii=False).encode()).hexdigest() def create(name, label, interface, address, host, *, mode='dhcp'): network = ipaddress.IPv4Interface(address).network return validate(dict(version=1,id=identity(name),label=label, network=dict(interface=interface,mode=mode,address=address if mode=='static' else '',gateway='',dns=[],ssh_port=1023), control_port=38443,redis_port=16379,redis_allow=[host],media=[],artifact_lan_cidr=str(network),selected_web='main', webs=[dict(id='main',label='주 연결 Web',host=host,manager='https://'+host,event_url='https://'+host+'/api/v1/events',server_id='AI-01')]))