"""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 import server_context as server from urllib.parse import urlsplit SETUP = server.path('setup', '/workspace/contextra/setup') PROFILES = SETUP / 'profiles' STATE = server.path('state', '/var/lib/contextra-maintenance/field') RUNTIME = server.path('runtime', '/workspace/contextra/runtime') def account(): return pwd.getpwnam(server.operator()) 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')]))