"""Korean field console. Configuration tools remain reachable when Runtime is blocked.""" import copy import fcntl import getpass import json import os from pathlib import Path import socket import subprocess import time from urllib.parse import urlsplit import zipfile import field_profiles as profiles import server_context as server from field_ui import print, input, text def ask(label, current='', *, clear=False): suffix=' / -=비우기' if clear else '' value=input(f'{label} [{current or "미설정"}; Enter=현재값 유지{suffix}]: ').strip() return '' if clear and value=='-' else value or current def choose_profile(): rows=sorted(profiles.PROFILES.glob('*.json')) for i,p in enumerate(rows,1): try: item=profiles.load(p.stem);print(i,item['id'],item['label']) except (ValueError,OSError):print(i,p.stem,'설정 오류') value=input('프로필 번호 / Enter=취소: ').strip() if not value:return None if not value.isdigit() or not 1<=int(value)<=len(rows):raise ValueError('번호 오류') return profiles.load(rows[int(value)-1].stem) def paths(p): server.banner() print('환경:',p['label']) print('Editable profile ('+server.operator()+'):',profiles.path(p['id'])) print('계정 파일(소유자 전용, 값 출력 없음):',profiles.secret_path(p)) print('선택 Web의 공개 CA:',profiles.ca_path(p)) print('NIC 실행 설정(root): /etc/netplan/*.yaml') print('앱 실행 site(root):',profiles.RUNTIME/'state/site.json') try: import connection print('앱 실행 Compose(root):',connection.active_compose()) site=profiles.read_json(profiles.RUNTIME/'state/site.json') print('기존 env 입력(root, 내용은 출력하지 않음):',*site['env_files'],sep='\n ') except Exception:print('실행 설정 조회 실패. 프로필 편집은 계속 사용할 수 있습니다.') print('설정 편집: ct profile edit / ct profile file') print('NIC 적용: ct net apply / 앱 준비: ct config prepare / 명시적 연결: ct config connect') print('기존 파일 변경 가져오기: ct emergency / 계정 저장: ct profile credentials / CA: ct ca') print('Editable source: network/webs/ports/redis_allow/media in the profile. Format version stays 1.') print('Connection revision: webs[].connection_version or ct config version NUMBER; then prepare.') print('Runtime site/Compose/env: supported imports only via ct emergency; do not edit active hashes or prepared files.') print('Credentials and CA: use ct profile credentials / ct ca; never publish private files.') def next_step(p): if not profiles.secret_path(p).is_file():step='ct profile credentials' elif not profiles.ca_path(p).is_file():step='ct ca' elif (profiles.STATE/'pending.json').exists():step='ct config status (preserve pending; inspect before retry/cancel)' else:step='ct config version status; ct config prepare' print('Next:',step) def edit(p): p=copy.deepcopy(p);n=p['network'] p['label']=ask('환경 이름',p['label']) n['interface']=ask('유선 NIC',n['interface']) n['mode']=ask('주소 방식 dhcp/static',n['mode']) n['address']=ask('고정 IPv4/prefix (DHCP에서는 비워도 됨)',n['address'],clear=True) n['gateway']=ask('Gateway',n['gateway'],clear=True) n['dns']=[x.strip() for x in ask('DNS 쉼표 목록',','.join(n['dns']),clear=True).split(',') if x.strip()] print('SSH listener unchanged:',p['network']['ssh_port'],'Listener changes are separate from NIC settings.') p['control_port']=int(ask('AI control/WSS 포트',str(p['control_port']))) p['redis_port']=int(ask('AI Redis 공개 포트',str(p['redis_port']))) p['artifact_lan_cidr']=ask('미디어 접근 허용 CIDR',p['artifact_lan_cidr']) p['redis_allow']=[x.strip() for x in ask('Redis 허용 Web IPv4 쉼표 목록',','.join(p['redis_allow'])).split(',') if x.strip()] w=profiles.web(p) w['host']=ask('선택 Web IPv4',w['host']) w['manager']=ask('Manager HTTPS origin (scheme/IP/port)',w['manager']) w['event_url']=ask('Event 전체 URL',w['event_url']) w['server_id']=ask('Web이 발급/기대하는 AI server ID',w['server_id']) print('로그인 경로는 기존 계약 /api/v1/auth/login 입니다. 다른 Web 후보·미디어 목록은 ct profile file에서 편집합니다.') profiles.save(p);print('저장 완료. 아직 NIC/Runtime/로그인은 변경하지 않았습니다.') def credentials(p): from auth_control import login_id path=profiles.secret_path(p) old=profiles.read_credentials(path) if path.exists() else {'loginId':'','password':''} login=login_id(ask('AI 연결용 login ID',old['loginId'])) password=getpass.getpass(text('비밀번호 [Enter=기존 유지, 숨김 입력]: ')) if password and password!=getpass.getpass(text('새 비밀번호 확인: ')):raise ValueError('비밀번호 확인 불일치') password=password or old['password'] if not password or any(x in password for x in ('\x00','\r','\n')) or len(password.encode())>8192:raise ValueError('유효한 비밀번호가 필요합니다.') path.parent.mkdir(mode=0o700,parents=True,exist_ok=True) if os.geteuid()==0: u=profiles.account();os.chown(path.parent,u.pw_uid,u.pw_gid) profiles.atomic(path,{'loginId':login,'password':password},user=True) print('계정 저장:',path,'로그인 요청은 하지 않았습니다. 다음: ct config prepare/connect') def status(p=None, *, probe=False): import field_runtime as runtime if p is None: try:p=profiles.load() except (ValueError,OSError) as e: print('프로필 조회 오류:',str(e)[:200]);print('ct profile list / ct profile file로 수정하세요.');return print('선택 환경:',p['label'],'(',p['id'],')') paths(p) selected=profiles.web(p) print('선택 Web 설정:',selected['manager'],'Event:',selected['event_url'],'AI ID:',selected['server_id']) print('프로필 포트: control/WSS',p['control_port'],'Redis',p['redis_port'],'SSH',p['network']['ssh_port']) print('Redis 허용 주소:',', '.join(p['redis_allow'])) try: import field_network print('실제 NIC IPv4:',field_network.addresses(p['network']['interface'])) except Exception:print('실제 NIC 조회 실패: ip -4 -br addr') try: import connection a=profiles.read_json(profiles.RUNTIME/'state/active.json',1048576) site=profiles.read_json(profiles.RUNTIME/'state/site.json') print('앱 설정 AI 주소:',site['ai_ip'],'transaction:',a['transaction']) actual=runtime.public_values(profiles.read_json(connection.active_compose(),8*1024**2)) print('앱 실행 설정:',json.dumps(actual,ensure_ascii=False)) if site['ai_ip'] not in field_network.addresses(p['network']['interface']): print('주소 불일치: 아직 새 환경을 앱에 적용하지 않았습니다. 다음: ct config prepare/connect/apply') if server.portable(): import compose_backend compose_backend.status() else: print('실행 설정 지문:', '일치' if runtime.sha(connection.active_compose())==a['compose_sha256'] and runtime.sha(profiles.RUNTIME/'state/site.json')==a['site_sha256'] else '수동 변경 감지 - ct emergency') except Exception:print('Runtime 설정 오류. ct 도구는 사용 가능합니다. ct emergency 또는 ct config restore로 확인하세요.') print('계정 등록:',profiles.secret_path(p).is_file(),'CA 준비:',profiles.ca_path(p).is_file()) for key in ('pending.json','apply-pending.json','network-pending.json'): if (profiles.STATE/key).exists():print('진행 중:',key,'ct config resume / ct net restore') try: stage=runtime.pending_stage() if (stage/'enroll-result.json').exists():print('마지막 발급 결과:',profiles.read_json(stage/'enroll-result.json')) except (ValueError,OSError):pass for w in p['webs']: parsed=urlsplit(w['manager']);result='미측정 (ct diagnose로 TCP만 확인)' if probe: try: with socket.create_connection((parsed.hostname,parsed.port or 443),timeout=1):pass result='TCP 연결됨; TLS/로그인/WSS는 별도 확인' except OSError:result='TCP 연결 실패; 상대 전원/주소/포트 확인' print('Web:',w['label'],'주 연결' if w['id']==p['selected_web'] else '보조 후보',result) for m in p['media']: result='참고용; 실제 입력은 Web rtspUrl' if probe: try: with socket.create_connection((m['host'],m['port']),timeout=1):pass result+=' / TCP 연결됨' except OSError:result+=' / TCP 연결 실패 (다른 입력은 차단하지 않음)' print('미디어:',m['label'],result) try: import auth_control live={'connected':None,'note':'not probed; user-led acceptance pending'} if server.portable() else auth_control.wss() print('현재 실행 Sidecar:',live) except Exception:print('Sidecar 상태 미확인. 로그인 요청은 하지 않았습니다.') import connection_versions connection_versions.show(p) next_step(p) if probe: from field_diagnostics import neighbor try:print('Kernel neighbor (not reboot persistence):',neighbor(p)) except (OSError,subprocess.SubprocessError):print('Neighbor lookup failed; other diagnosis remains available.') def profile_action(args): action=args[0] if args else 'list' if action=='list': for path in sorted(profiles.PROFILES.glob('*.json')): try: p=profiles.load(path.stem);print(p['id'],p['label'],path) except (ValueError,OSError):print(path.stem,'설정 오류:',path) return if action=='select': p=profiles.load(args[1]) if len(args)>1 else choose_profile() if p: profiles.atomic(profiles.STATE/'selected.json',{'profile':p['id']}) print('선택만 변경했습니다. NIC/서비스/로그인 변경 없음.');paths(p) return if action=='create': base=profiles.load();name=profiles.identity(ask('새 프로필 이름(영문)')) if profiles.path(name).exists():raise ValueError('이미 있는 프로필입니다.') base['id']=name;base['label']=ask('환경 설명',name) profiles.save(base);print('일반 설정만 복사했습니다. 계정/CA는 새로 등록합니다.');return if action=='file': selected=profiles.read_json(profiles.STATE/'selected.json')['profile'] target=profiles.path(selected) print('편집 경로:',target) u=profiles.account() subprocess.run(['runuser','-u',u.pw_name,'--','nano','--',str(target)],check=True,timeout=3600) profiles.load();print('문법 검사 완료. 적용은 ct config prepare/connect 또는 ct net apply.') return p=profiles.load() if action=='edit':edit(p) elif action=='paths':paths(p) elif action=='credentials':credentials(p) elif action=='web': for i,w in enumerate(p['webs'],1):print(i,w['id'],w['label'],w['manager']) v=input('주 연결 Web 번호: ').strip() if not v.isdigit() or not 1<=int(v)<=len(p['webs']):raise ValueError('번호 오류') p['selected_web']=p['webs'][int(v)-1]['id'];profiles.save(p) print('주 Web 선택만 저장했습니다. 계정/CA는 Web별로 별도입니다.') elif action=='delete': if len(args)!=2:raise ValueError('ct profile delete 이름') name=profiles.identity(args[1]) if name==p['id']:raise ValueError('선택 중인 프로필은 삭제하지 않습니다.') if input('프로필을 보존 archive로 이동하려면 DELETE 입력: ')!='DELETE':return archive=profiles.safe_path(profiles.PROFILES/'archive');archive.mkdir(exist_ok=True,mode=0o700) target=profiles.safe_path(archive/(name+'-'+time.strftime('%Y%m%dT%H%M%S')+'.json')) if target.exists():raise ValueError('같은 archive가 있습니다.') profiles.path(name).rename(target);print('archive:',target,'계정/CA 원본은 보존했습니다.') else:raise ValueError('profile: list/select/create/edit/file/paths/web/credentials/delete') def config_action(args): import field_runtime as runtime action=args[0] if args else 'status' if action=='status':status();return if action=='version': import connection_versions connection_versions.configure(args[1:]) elif action=='prepare': p=profiles.load();paths(p) print('선택 프로필을 새 준비본으로 만듭니다. 아직 로그인/기존 설정 적용은 하지 않습니다.') if input('PREPARE 입력: ')=='PREPARE':runtime.prepare(p) elif action in ('connect','retry'): import network_prepare stage=runtime.pending_stage() if (profiles.STATE/'pending.json').exists() else runtime.prepare(profiles.load()) record,p=runtime.verify_pending(stage) print('환경:',p['label'],'Web:',profiles.web(p)['manager']) body=profiles.read_json(stage/'bridge/login-body.json') print('Sending connectionConfigVersion:',body.get('connectionConfigVersion'),'Web current version: UNKNOWN') print('선택 프로필의 계정으로 로그인/인증서 요청1회. 기존 token이 폐기될 수 있습니다.') print('카메라/서비스가 실행 중이면 먼저 ct stop이 필요합니다. 성공 후에도 자동 Start하지 않습니다.') intent=stage/'enroll-intent.json' archive_intent=False if intent.exists(): try:runtime.stage_ready(stage) except Exception: if action!='retry':raise ValueError('이전 요청 결과 미확정. 원격 처리 확인 후 ct config retry를 명시적으로 사용하세요.') if input('이전 원격 결과를 확인했고 새 요청을 승인하려면 RETRY 입력: ')!='RETRY':return archive_intent=True if input('명시적 연결 요청을 하려면 CONNECT 입력: ')!='CONNECT':return if archive_intent:intent.rename(stage/('enroll-intent-'+str(time.time_ns())+'.previous.json')) try:network_prepare.enroll(stage) except Exception: from field_diagnostics import explain_result explain_result(profiles.read_json(stage/'enroll-result.json')) raise print('발급 확인. 로컬 설정 적용: ct config apply') elif action in ('apply','resume'): stage=runtime.pending_stage() if input('검증된 준비본을 적용하고 서비스 정지 상태로 유지하려면 APPLY 입력: ')=='APPLY':runtime.apply_stage(stage) elif action=='restore': if input('로컬 설정만 이전 기준으로 복구하려면 RESTORE 입력: ')=='RESTORE':runtime.restore_config() elif action=='cancel': stage=runtime.pending_stage() if (profiles.STATE/'apply-pending.json').exists():raise ValueError('적용 중입니다. resume/restore를 사용하세요.') if (stage/'tls/control-tls.crt').exists() or any((stage/'bridge-session').glob('*')): raise ValueError('Session/certificate exists. Reconcile the remote result before cancelling; evidence preserved.') result=profiles.read_json(stage/'enroll-result.json') if (stage/'enroll-result.json').exists() else {} if (stage/'enroll-intent.json').exists() and result.get('error')!='LOGIN_REJECTED':raise ValueError('원격 요청 이력이 있습니다. 발급/처리 상태 확인 후 resume/retry하세요.') if input('미발급 준비본을 보존하고 선택 해제하려면 CANCEL 입력: ')=='CANCEL': (profiles.STATE/'pending.json').unlink();print('기존 Runtime 유지. 준비본은 보존했습니다.') elif action in ('export','note','confirm'): last=profiles.read_json(profiles.STATE/'last-applied.json') stage=profiles.safe_path(last['stage']) out=profiles.SETUP/'outbox'/('field-change-'+stage.name.removeprefix('field-')) if action=='note': note=input('비밀정보 없는 수동 작업 메모: ') if len(note)>2000:raise ValueError('메모2000자 상한') summary=profiles.safe_path(out/'SUMMARY.ko.md') profiles.atomic(summary,summary.read_bytes()+('\n- 수동 메모: '+note+'\n').encode(),user=True) elif action=='confirm': print('실제로 확인한 항목만 적으세요. 예: 로그인/WSS/Redis/카메라1 Start-Stop. 미확인은 제외.') confirmed=input('사용자 확인 결과: ').strip() if not confirmed or len(confirmed)>1000:raise ValueError('확인 결과1..1000자') v=profiles.read_json(out/'verification.json');v['user_observation']=confirmed;v['web_acceptance']='USER_REPORTED' profiles.atomic(out/'verification.json',v,user=True) print('사용자 관측으로 기록했습니다. 자동 검증 성공으로 대체하지 않습니다.') else: if input('메모 포함 비밀정보가 없는지 검토했다면 EXPORT 입력: ')!='EXPORT':return output=out.with_name(out.name+'-'+time.strftime('%H%M%S')+'.zip') profiles.safe_path(output) fd=os.open(output,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600) with os.fdopen(fd,'wb') as stream,zipfile.ZipFile(stream,'w',compression=zipfile.ZIP_DEFLATED) as archive: for name in ('SUMMARY.ko.md','changes.redacted.json','verification.json'): path=profiles.safe_path(out/name) if path.stat().st_size>1048576:raise ValueError('반출 크기 초과') archive.write(path,name) u=profiles.account();os.chown(output,u.pw_uid,u.pw_gid);output.chmod(0o600) print('USB 반출 파일:',output) else:raise ValueError('config: status/version/prepare/connect/apply/resume/retry/cancel/restore/note/confirm/export') def emergency(): import field_runtime as runtime p=profiles.load();paths(p) print('SHA 불일치 일반 설정 가져오기. image/코드/디스크/인증 검증은 끄지 않습니다.') print('1 실행 Compose 2 기존 env/overlay 재해석값 3 site AI 주소 Enter=취소') sources={'1':'compose','2':'persistent','3':'site'} chosen=sources.get(input('원본 선택: ').strip()) if not chosen:return proposed=runtime.import_changes(p,chosen) print(json.dumps(proposed,ensure_ascii=False,indent=2)) print('프로필 값과 충돌하면 위 가져온 값으로 바뀝니다. 서비스/로그인은 아직 변경하지 않습니다.') if input('프로필에 수용하려면 IMPORT 입력: ')=='IMPORT': profiles.save(proposed) print('수용 완료. 계정/CA는 별도로 확인하고 ct config prepare/connect/apply로 이어갑니다.') def main(action,args): profiles.STATE.mkdir(parents=True,exist_ok=True,mode=0o700) with (profiles.STATE/'console.lock').open('a') as lock: try:fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB) except BlockingIOError:raise ValueError('다른 ct 설정 작업이 진행 중입니다. 해당 터미널에서 완료/취소 후 재개하세요.') _main(action,args) def _main(action,args): if action=='profile':profile_action(args) elif action=='config':config_action(args) elif action=='emergency':emergency() elif action=='diagnose':status(probe=True) elif action=='ca': import field_ca field_ca.import_ca(profiles.load(),args[0] if args else None) elif action=='net': server.require('network') import field_network op=args[0] if args else 'status' if op=='status':field_network.status(profiles.load()) elif op=='arp': import field_diagnostics field_diagnostics.arp(args[1:]) elif op=='apply':field_network.apply(profiles.load()) elif op in ('nextboot','cancel','scheduled'): import network_schedule network_schedule.main(op,profiles.load()) elif op=='restore': if input('이전 NIC 설정으로 복구하려면 RESTORE 입력: ')=='RESTORE':field_network.restore() else:raise ValueError('net: status/apply/restore')