962 lines
42 KiB
Python
962 lines
42 KiB
Python
#!/usr/bin/env python3
|
||
"""Независимый решатель задачи 5 «Мы снова пришли к носу» (CATD, версия 2:
|
||
язык сцены).
|
||
|
||
Вход: участнический стартовый PNG (argv[1]).
|
||
python solvers/solve_task5.py dist/participants/kot5_nos.png [--render cat.png]
|
||
|
||
Конвейер:
|
||
PNG чанки -> caTs -> CCAP (version, uncompressed_length, sha256) ->
|
||
zlib (лимит uncompressed_length и жёсткий потолок 64 МиБ) -> PCAPNG
|
||
(Section/Interface/Enhanced Packet) -> TCP-потоки по соединениям
|
||
(sequence numbers, ретрансмиты, сегментация) -> HTTP POST /upload
|
||
(Content-Length) -> тела PNG -> дескриптор CDSC (tEXt CATD.desc) ->
|
||
калибровка 01010101 -> биты (пары плиток 4x4, знак разности средних
|
||
S=R+G, MSB-first) -> пакеты CATD (magic/version/crc32) ->
|
||
фильтр session_id (мода) -> удаление точных дублей / отказ при
|
||
конфликтующих -> полнота по total_packets -> сортировка по
|
||
(session_id, sequence) -> проверка END SHA-256 -> поток команд языка
|
||
сцены (спека — в START payload) -> исполнение сцены: шаблоны, группы,
|
||
DELETE в момент чтения, единственный PUBLISH, композиция преобразований
|
||
(своя группа -> родительская -> ... -> корень) -> MARK id=1 в
|
||
опубликованной сцене (единственный) -> мировые координаты носа ->
|
||
публичная функция флага -> флаг.
|
||
|
||
Зависимости: stdlib + Pillow + numpy. Генератор не импортируется,
|
||
файлы организатора не читаются.
|
||
|
||
Правило фильтрации сессий: session_id выбирается как самый частый
|
||
(мода) среди валидных пакетов; пакеты с другими session_id
|
||
отбрасываются; при равенстве частот берётся первый встреченный.
|
||
"""
|
||
import argparse
|
||
import hashlib
|
||
import io
|
||
import struct
|
||
import sys
|
||
import zlib
|
||
from collections import Counter
|
||
|
||
import numpy as np
|
||
from PIL import Image, ImageDraw
|
||
|
||
# --- Лимиты безопасности -------------------------------------------------
|
||
MAX_CCAP_COMPRESSED = 64 * 1024 * 1024 # сжатый PCAPNG
|
||
MAX_CCAP_UNCOMPRESSED = 64 * 1024 * 1024 # жёсткий потолок распаковки
|
||
MAX_PCAPNG_PACKETS = 1_000_000
|
||
MAX_HTTP_BODY = 32 * 1024 * 1024
|
||
MAX_PNG_BODIES = 100_000
|
||
MAX_PACKET_PAYLOAD = 4096
|
||
MAX_POSITIONS = 8192
|
||
MAX_DESCRIPTOR_COMPRESSED = 32 * 1024 # v2: позиций больше (START со спекой сцены)
|
||
|
||
CALIBRATION_BYTE = 0x55
|
||
PAIR_W, PAIR_H, TILE_W = 8, 4, 4
|
||
|
||
TYPE_START, TYPE_DATA, TYPE_END = 1, 2, 4 # ACK=3 зарезервирован, не используется
|
||
|
||
# Язык сцены (спека публикуется в START payload).
|
||
(OP_DEF_TEMPLATE, OP_END_DEF, OP_LINE, OP_MARK, OP_GROUP, OP_END_GROUP,
|
||
OP_TRANSLATE, OP_ROT90, OP_REFLECT, OP_INSTANCE, OP_DELETE,
|
||
OP_PUBLISH) = range(1, 13)
|
||
|
||
FLAG_MODULUS = 2000001
|
||
|
||
|
||
class SolveError(Exception):
|
||
"""Безопасная ошибка решателя (без traceback для ожидаемых отказов)."""
|
||
|
||
|
||
# --- PNG чанки ------------------------------------------------------------
|
||
PNG_SIG = b"\x89PNG\r\n\x1a\n"
|
||
|
||
|
||
def parse_png_chunks(data):
|
||
if len(data) < 8 or data[:8] != PNG_SIG:
|
||
raise SolveError("не PNG (сигнатура)")
|
||
chunks = []
|
||
off = 8
|
||
while off < len(data):
|
||
if len(data) - off < 12:
|
||
raise SolveError("усечённый чанк PNG")
|
||
(length,) = struct.unpack_from(">I", data, off)
|
||
if length > len(data) - off - 12:
|
||
raise SolveError("длина чанка PNG превышает остаток файла")
|
||
ctype = data[off + 4:off + 8]
|
||
body = data[off + 8:off + 8 + length]
|
||
(want_crc,) = struct.unpack_from(">I", data, off + 8 + length)
|
||
if zlib.crc32(ctype + body) & 0xFFFFFFFF != want_crc:
|
||
raise SolveError(f"crc чанка {ctype!r} не сошёлся")
|
||
chunks.append((ctype, body))
|
||
off += 12 + length
|
||
if ctype == b"IEND":
|
||
break
|
||
if not chunks or chunks[-1][0] != b"IEND":
|
||
raise SolveError("нет IEND")
|
||
return chunks
|
||
|
||
|
||
def png_text_value(chunks, keyword):
|
||
prefix = keyword.encode("latin-1") + b"\x00"
|
||
for ctype, body in chunks:
|
||
if ctype == b"tEXt" and body.startswith(prefix):
|
||
return body[len(prefix):].decode("latin-1")
|
||
return None
|
||
|
||
|
||
# --- CCAP (чанк caTs) ------------------------------------------------------
|
||
def extract_pcapng(png_bytes):
|
||
chunks = parse_png_chunks(png_bytes)
|
||
cats = [body for ctype, body in chunks if ctype == b"caTs"]
|
||
if not cats:
|
||
raise SolveError("в стартовом PNG нет чанка caTs с захватом")
|
||
body = cats[0]
|
||
if len(body) < 41:
|
||
raise SolveError("CCAP слишком короткий")
|
||
if body[:4] != b"CCAP":
|
||
raise SolveError("неверный magic CCAP")
|
||
if body[4] != 1:
|
||
raise SolveError(f"неподдерживаемая версия CCAP: {body[4]}")
|
||
(ulen,) = struct.unpack_from("<I", body, 5)
|
||
if ulen > MAX_CCAP_UNCOMPRESSED:
|
||
raise SolveError(f"CCAP uncompressed_length {ulen} превышает 64 МиБ")
|
||
want_sha = body[9:41]
|
||
compressed = body[41:]
|
||
if len(compressed) > MAX_CCAP_COMPRESSED:
|
||
raise SolveError("CCAP сжатые данные превышают 64 МиБ")
|
||
d = zlib.decompressobj()
|
||
out = d.decompress(compressed, min(ulen, MAX_CCAP_UNCOMPRESSED) + 1)
|
||
if len(out) > MAX_CCAP_UNCOMPRESSED or len(out) > ulen:
|
||
raise SolveError("CCAP распаковался больше лимита")
|
||
if len(out) != ulen:
|
||
raise SolveError(f"CCAP длина {len(out)} != uncompressed_length {ulen}")
|
||
if hashlib.sha256(out).digest() != want_sha:
|
||
raise SolveError("CCAP sha256 распакованного PCAPNG не сошёлся")
|
||
return out
|
||
|
||
|
||
# --- PCAPNG -----------------------------------------------------------------
|
||
def parse_pcapng(data):
|
||
"""Минимальный разбор PCAPNG: SHB, IDB, EPB (+SPB). Без внешних библиотек."""
|
||
interfaces = [] # linktype по индексу интерфейса
|
||
packets = [] # (ts_raw, iface_id, bytes)
|
||
off = 0
|
||
bo = "<" # порядок байт секции
|
||
while off < len(data):
|
||
if len(data) - off < 12:
|
||
raise SolveError("PCAPNG: усечённый блок")
|
||
(btype,) = struct.unpack_from("<I", data, off)
|
||
if btype == 0x0A0D0D0A: # Section Header
|
||
(blen,) = struct.unpack_from("<I", data, off + 4)
|
||
if blen < 28 or off + blen > len(data):
|
||
raise SolveError("PCAPNG: неверная длина SHB")
|
||
(bom,) = struct.unpack_from("<I", data, off + 8)
|
||
if bom == 0x1A2B3C4D:
|
||
bo = "<"
|
||
elif bom == 0x4D3C2B1A:
|
||
bo = ">"
|
||
else:
|
||
raise SolveError("PCAPNG: неверный byte-order magic")
|
||
interfaces = []
|
||
else:
|
||
(blen,) = struct.unpack_from(bo + "I", data, off + 4)
|
||
if blen < 12 or off + blen > len(data):
|
||
raise SolveError("PCAPNG: неверная длина блока")
|
||
if btype == 1: # Interface Description
|
||
if blen < 20:
|
||
raise SolveError("PCAPNG: короткий IDB")
|
||
(linktype,) = struct.unpack_from(bo + "H", data, off + 8)
|
||
interfaces.append(linktype)
|
||
elif btype == 6: # Enhanced Packet
|
||
if blen < 32:
|
||
raise SolveError("PCAPNG: короткий EPB")
|
||
iface_id, ts_hi, ts_lo, caplen, _origlen = struct.unpack_from(
|
||
bo + "IIIII", data, off + 8)
|
||
if caplen > blen - 32:
|
||
raise SolveError("PCAPNG: caplen превышает блок")
|
||
pkt = data[off + 28:off + 28 + caplen]
|
||
packets.append(((ts_hi << 32) | ts_lo, iface_id, bytes(pkt)))
|
||
if len(packets) > MAX_PCAPNG_PACKETS:
|
||
raise SolveError("PCAPNG: слишком много пакетов")
|
||
elif btype == 3: # Simple Packet
|
||
if blen < 16:
|
||
raise SolveError("PCAPNG: короткий SPB")
|
||
(origlen,) = struct.unpack_from(bo + "I", data, off + 8)
|
||
caplen = min(origlen, blen - 16)
|
||
pkt = data[off + 12:off + 12 + caplen]
|
||
packets.append((0, 0, bytes(pkt)))
|
||
# остальные блоки (Name Resolution и пр.) пропускаем
|
||
off += blen
|
||
if not interfaces:
|
||
raise SolveError("PCAPNG: нет Interface Description Block")
|
||
packets.sort(key=lambda p: p[0]) # порядок по временным меткам
|
||
return interfaces, packets
|
||
|
||
|
||
# --- TCP --------------------------------------------------------------------
|
||
def parse_ipv4_tcp(pkt, linktype):
|
||
"""Возвращает (src, sport, dst, dport, seq, flags, payload) или None."""
|
||
if linktype == 1: # Ethernet
|
||
if len(pkt) < 14:
|
||
return None
|
||
(ethertype,) = struct.unpack_from(">H", pkt, 12)
|
||
if ethertype == 0x8100: # VLAN
|
||
if len(pkt) < 18:
|
||
return None
|
||
(ethertype,) = struct.unpack_from(">H", pkt, 16)
|
||
ip = pkt[18:]
|
||
else:
|
||
ip = pkt[14:]
|
||
if ethertype != 0x0800:
|
||
return None
|
||
elif linktype in (101, 228): # raw IP
|
||
ip = pkt
|
||
else:
|
||
return None
|
||
if len(ip) < 20 or ip[0] >> 4 != 4:
|
||
return None
|
||
ihl = (ip[0] & 0x0F) * 4
|
||
if ihl < 20 or len(ip) < ihl:
|
||
return None
|
||
(total_len,) = struct.unpack_from(">H", ip, 2)
|
||
if total_len < ihl or total_len > len(ip):
|
||
return None
|
||
if ip[9] != 6: # TCP
|
||
return None
|
||
src = ".".join(str(b) for b in ip[12:16])
|
||
dst = ".".join(str(b) for b in ip[16:20])
|
||
tcp = ip[ihl:total_len]
|
||
if len(tcp) < 20:
|
||
return None
|
||
sport, dport, seq, _ack, off_flags = struct.unpack_from(">HHIIB", tcp, 0)
|
||
data_off = (off_flags >> 4) * 4
|
||
if data_off < 20 or data_off > len(tcp):
|
||
return None
|
||
flags = tcp[13]
|
||
payload = tcp[data_off:]
|
||
return src, sport, dst, dport, seq, flags, payload
|
||
|
||
|
||
def reassemble_stream(segments, isn):
|
||
"""Сборка потока из сегментов (seq, payload). Ретрансмиты с совпадающими
|
||
байтами пропускаются; конфликт и дыра — ошибки."""
|
||
segments = sorted(segments)
|
||
buf = bytearray()
|
||
next_seq = isn
|
||
for seq, payload in segments:
|
||
if seq > next_seq:
|
||
raise SolveError(f"TCP: дыра в потоке (seq {seq}, ждали {next_seq})")
|
||
if seq < isn:
|
||
raise SolveError("TCP: сегмент начинается до начала потока")
|
||
skip = next_seq - seq
|
||
overlap_len = min(skip, len(payload))
|
||
start = seq - isn
|
||
if payload[:overlap_len] != bytes(buf[start:start + overlap_len]):
|
||
raise SolveError("TCP: конфликтующий ретрансмит")
|
||
if skip >= len(payload):
|
||
continue
|
||
buf.extend(payload[skip:])
|
||
next_seq += len(payload) - skip
|
||
return bytes(buf)
|
||
|
||
|
||
def extract_http_png_bodies(interfaces, packets):
|
||
"""TCP-потоки по соединениям -> HTTP -> тела PNG POST /upload.
|
||
|
||
Keep-alive: несколько POST в одном client->server потоке. Посторонние
|
||
потоки (шумовые GET к другим сервисам) игнорируются. Ретрансмиты и
|
||
out-of-order (netem loss/reorder) переживаются сборкой по seq с
|
||
детектом конфликтов. Interim-ответы 100 Continue пропускаются при
|
||
разборе server->server направления (count_http_responses)."""
|
||
flows = {} # (src,sport,dst,dport) -> {"syn": isn, "segs": [(seq,payload)]}
|
||
for _ts, iface_id, pkt in packets:
|
||
if iface_id >= len(interfaces):
|
||
continue
|
||
parsed = parse_ipv4_tcp(pkt, interfaces[iface_id])
|
||
if parsed is None:
|
||
continue
|
||
src, sport, dst, dport, seq, flags, payload = parsed
|
||
key = (src, sport, dst, dport)
|
||
f = flows.setdefault(key, {"syn": None, "segs": []})
|
||
if flags & 0x02: # SYN
|
||
f["syn"] = seq
|
||
if payload:
|
||
f["segs"].append((seq, payload))
|
||
bodies = []
|
||
responses = 0
|
||
for (src, sport, dst, dport), f in flows.items():
|
||
if not f["segs"]:
|
||
continue
|
||
if f["syn"] is None:
|
||
raise SolveError(f"TCP: нет SYN для {src}:{sport}->{dst}:{dport} "
|
||
"(захват неполный)")
|
||
stream = reassemble_stream(f["segs"], f["syn"] + 1)
|
||
if stream.startswith(b"HTTP/"):
|
||
# Направление ответов: считаем финальные, пропуская interim 1xx.
|
||
responses += count_http_responses(stream)
|
||
continue
|
||
first = stream.split(b"\r\n", 1)[0]
|
||
parts = first.split(b" ")
|
||
if len(parts) != 3 or not parts[0].isalpha() or not parts[0].isupper() \
|
||
or not parts[2].startswith(b"HTTP/"):
|
||
continue # не HTTP-запросы — чужой трафик
|
||
bodies.extend(parse_http_posts(stream))
|
||
if responses:
|
||
print(f"solver: финальных HTTP-ответов в захвате: {responses}", file=sys.stderr)
|
||
if len(bodies) > MAX_PNG_BODIES:
|
||
raise SolveError("слишком много PNG-тел")
|
||
return bodies
|
||
|
||
|
||
def _parse_head(stream, off):
|
||
"""Заголовки HTTP-сообщения (запроса или ответа) с позиции off."""
|
||
end = stream.find(b"\r\n\r\n", off)
|
||
if end < 0:
|
||
raise SolveError("HTTP: усечённые заголовки")
|
||
lines = stream[off:end].decode("latin-1").split("\r\n")
|
||
headers = {}
|
||
for line in lines[1:]:
|
||
k, _, v = line.partition(":")
|
||
headers[k.strip().lower()] = v.strip()
|
||
return lines[0], headers, end + 4
|
||
|
||
|
||
def _skip_chunked(stream, off):
|
||
"""Пропуск chunked-тела; возвращает позицию после него или None."""
|
||
while True:
|
||
eol = stream.find(b"\r\n", off)
|
||
if eol < 0:
|
||
return None
|
||
try:
|
||
size = int(stream[off:eol].split(b";")[0], 16)
|
||
except ValueError:
|
||
return None
|
||
if size < 0 or size > MAX_HTTP_BODY:
|
||
return None
|
||
off = eol + 2
|
||
if size == 0:
|
||
# После last-chunk идут трейлеры и пустая строка.
|
||
if stream[off:off + 2] == b"\r\n":
|
||
return off + 2
|
||
end = stream.find(b"\r\n\r\n", off)
|
||
return None if end < 0 else end + 4
|
||
if off + size + 2 > len(stream):
|
||
return None
|
||
if stream[off + size:off + size + 2] != b"\r\n":
|
||
return None
|
||
off += size + 2
|
||
|
||
|
||
def count_http_responses(stream):
|
||
"""Разбор server->client потока: interim-ответы (1xx, в т.ч.
|
||
'HTTP/1.1 100 Continue') пропускаются; тела финальных ответов
|
||
пропускаются по Content-Length или chunked. Возвращает число
|
||
финальных ответов."""
|
||
n, off = 0, 0
|
||
while off < len(stream):
|
||
try:
|
||
first, headers, body_start = _parse_head(stream, off)
|
||
except SolveError:
|
||
break
|
||
parts = first.split(" ")
|
||
if len(parts) < 2 or not parts[0].startswith("HTTP/"):
|
||
break
|
||
try:
|
||
status = int(parts[1])
|
||
except ValueError:
|
||
break
|
||
if 100 <= status < 200:
|
||
off = body_start # interim-ответ без тела — пропускаем
|
||
continue
|
||
if headers.get("transfer-encoding", "").lower() == "chunked":
|
||
nxt = _skip_chunked(stream, body_start)
|
||
if nxt is None:
|
||
break
|
||
off = nxt
|
||
else:
|
||
try:
|
||
clen = int(headers.get("content-length", "0"))
|
||
except ValueError:
|
||
break
|
||
if clen < 0 or clen > MAX_HTTP_BODY:
|
||
break
|
||
off = body_start + clen
|
||
n += 1
|
||
return n
|
||
|
||
|
||
def parse_http_posts(stream):
|
||
"""Разбор последовательности HTTP-запросов в client->server потоке
|
||
(keep-alive: несколько подряд). Забирает только POST /upload с
|
||
Content-Type: image/png; посторонние запросы (шумовые GET и пр.)
|
||
пропускаются с разбором их длины. Заголовок Expect: 100-continue на
|
||
разбор не влияет (interim-ответ живёт в другом направлении)."""
|
||
bodies = []
|
||
off = 0
|
||
while off < len(stream):
|
||
first, headers, body_start = _parse_head(stream, off)
|
||
parts = first.split(" ")
|
||
if len(parts) != 3:
|
||
raise SolveError(f"HTTP: ожидался запрос, получено {first!r}")
|
||
method, path = parts[0], parts[1]
|
||
clen = 0
|
||
if "content-length" in headers:
|
||
try:
|
||
clen = int(headers["content-length"])
|
||
except ValueError:
|
||
raise SolveError("HTTP: битый Content-Length")
|
||
if clen < 0 or clen > MAX_HTTP_BODY:
|
||
raise SolveError(f"HTTP: Content-Length {clen} вне лимита")
|
||
if body_start + clen > len(stream):
|
||
raise SolveError("HTTP: тело усечено (не хватило сегментов)")
|
||
body = stream[body_start:body_start + clen]
|
||
off = body_start + clen
|
||
if method != "POST" or path != "/upload" or \
|
||
headers.get("content-type") != "image/png":
|
||
print(f"solver: пропущен посторонний запрос {method} {path}",
|
||
file=sys.stderr)
|
||
continue
|
||
if not body.startswith(PNG_SIG):
|
||
raise SolveError("HTTP: тело не PNG")
|
||
bodies.append(body)
|
||
return bodies
|
||
|
||
|
||
# --- Дескриптор CDSC ---------------------------------------------------------
|
||
def decode_descriptor(chunks):
|
||
text = png_text_value(chunks, "CATD.desc")
|
||
if text is None:
|
||
raise SolveError("в сетевом PNG нет tEXt CATD.desc")
|
||
import base64
|
||
try:
|
||
raw = base64.b64decode(text, validate=True)
|
||
except Exception as e:
|
||
raise SolveError(f"CDSC base64: {e}")
|
||
if len(raw) > MAX_DESCRIPTOR_COMPRESSED:
|
||
raise SolveError("CDSC сжатый дескриптор превышает 32 КиБ")
|
||
try:
|
||
d = zlib.decompressobj()
|
||
data = d.decompress(raw, 21 + 4 * MAX_POSITIONS + 4 + 1)
|
||
except zlib.error as e:
|
||
raise SolveError(f"CDSC zlib: {e}")
|
||
if len(data) > 21 + 4 * MAX_POSITIONS + 4:
|
||
raise SolveError("CDSC распакованный дескриптор превышает лимит")
|
||
return parse_descriptor(data)
|
||
|
||
|
||
def parse_descriptor(b):
|
||
if len(b) < 25:
|
||
raise SolveError("CDSC слишком короткий")
|
||
if b[:4] != b"CDSC":
|
||
raise SolveError("CDSC неверный magic")
|
||
if b[4] != 1:
|
||
raise SolveError(f"CDSC неподдерживаемая версия {b[4]}")
|
||
width, height, blocks_across, count = struct.unpack_from("<IIII", b, 5)
|
||
if count == 0 or count > MAX_POSITIONS:
|
||
raise SolveError(f"CDSC недопустимое число позиций {count}")
|
||
if len(b) != 21 + 4 * count + 4:
|
||
raise SolveError("CDSC длина не согласована с count")
|
||
(want_crc,) = struct.unpack_from("<I", b, len(b) - 4)
|
||
if zlib.crc32(b[:len(b) - 4]) & 0xFFFFFFFF != want_crc:
|
||
raise SolveError("CDSC crc32 не сошёлся")
|
||
positions = list(struct.unpack_from(f"<{count}I", b, 21))
|
||
total = blocks_across * (height // PAIR_H)
|
||
prev = -1
|
||
for p in positions:
|
||
if p >= total:
|
||
raise SolveError("CDSC позиция вне сетки")
|
||
if p <= prev:
|
||
raise SolveError("CDSC позиции не отсортированы строго")
|
||
prev = p
|
||
return width, height, blocks_across, positions
|
||
|
||
|
||
# --- Пиксельный канал ---------------------------------------------------------
|
||
class BitReader:
|
||
"""Читатель бит: 1 — левая плитка светлее (diff>0), 0 — темнее.
|
||
|diff| < 1.5 (|sumDiff| < 24) — ошибка «сомнительный бит»."""
|
||
|
||
def __init__(self, img, blocks_across, positions):
|
||
self.s = img[:, :, 0].astype(np.int64) + img[:, :, 1].astype(np.int64)
|
||
self.ba = blocks_across
|
||
self.pos = positions
|
||
self.k = 0
|
||
|
||
def bit(self):
|
||
if self.k >= len(self.pos):
|
||
raise SolveError("позиции носителя исчерпаны")
|
||
idx = self.pos[self.k]
|
||
self.k += 1
|
||
r, c = divmod(idx, self.ba)
|
||
y0, x0 = PAIR_H * r, PAIR_W * c
|
||
tile_l = self.s[y0:y0 + PAIR_H, x0:x0 + TILE_W]
|
||
tile_r = self.s[y0:y0 + PAIR_H, x0 + TILE_W:x0 + PAIR_W]
|
||
d = int(tile_l.sum()) - int(tile_r.sum())
|
||
if d >= 24:
|
||
return 1
|
||
if d <= -24:
|
||
return 0
|
||
raise SolveError("сомнительный бит (|diff| < 1.5)")
|
||
|
||
def byte(self):
|
||
v = 0
|
||
for _ in range(8):
|
||
v = (v << 1) | self.bit()
|
||
return v
|
||
|
||
def read(self, n):
|
||
return bytes(self.byte() for _ in range(n))
|
||
|
||
|
||
def check_carrier_invariants(img, blocks_across, positions, used):
|
||
"""Проверка инвариантов, действительно неизменных при кодировании:
|
||
размах (R-G) по 32 пикселям пары <= 4. Диапазон RGB и разность средних
|
||
кодированием намеренно изменяются (δ ∈ {-2..2}), поэтому декодер их
|
||
не проверяет (см. SPECS.md 5.1: отбор — по оригиналу генератором)."""
|
||
rg = img[:, :, 0].astype(np.int32) - img[:, :, 1].astype(np.int32)
|
||
for idx in positions[:used]:
|
||
r, c = divmod(idx, blocks_across)
|
||
y0, x0 = PAIR_H * r, PAIR_W * c
|
||
blk = rg[y0:y0 + PAIR_H, x0:x0 + PAIR_W]
|
||
if int(blk.max()) - int(blk.min()) > 4:
|
||
raise SolveError(f"позиция {idx}: инвариант R-G нарушен")
|
||
|
||
|
||
def decode_packet(png_bytes):
|
||
chunks = parse_png_chunks(png_bytes)
|
||
width, height, blocks_across, positions = decode_descriptor(chunks)
|
||
img = np.asarray(Image.open(io.BytesIO(png_bytes)).convert("RGB"))
|
||
if img.shape[1] != width or img.shape[0] != height:
|
||
raise SolveError("размер PNG не совпадает с дескриптором")
|
||
br = BitReader(img, blocks_across, positions)
|
||
calib = br.byte()
|
||
if calib != CALIBRATION_BYTE:
|
||
raise SolveError(f"неверная калибровка: {calib:#04x}, ожидалось 0x55")
|
||
head = br.read(20)
|
||
if head[:4] != b"CATD":
|
||
raise SolveError("неверный magic пакета CATD")
|
||
if head[4] != 1:
|
||
raise SolveError(f"неподдерживаемая версия пакета: {head[4]}")
|
||
(plen,) = struct.unpack_from("<H", head, 18)
|
||
if plen > MAX_PACKET_PAYLOAD:
|
||
raise SolveError(f"payload_length {plen} превышает лимит")
|
||
rest = br.read(plen + 4)
|
||
raw = head + rest
|
||
(want_crc,) = struct.unpack_from("<I", raw, len(raw) - 4)
|
||
if zlib.crc32(raw[:len(raw) - 4]) & 0xFFFFFFFF != want_crc:
|
||
raise SolveError("crc32 пакета CATD не сошёлся")
|
||
ptype = head[5]
|
||
if ptype not in (TYPE_START, TYPE_DATA, TYPE_END):
|
||
raise SolveError(f"неизвестный тип пакета: {ptype}")
|
||
check_carrier_invariants(img, blocks_across, positions, br.k)
|
||
return {
|
||
"type": ptype,
|
||
"session_id": struct.unpack_from("<I", head, 6)[0],
|
||
"sequence": struct.unpack_from("<I", head, 10)[0],
|
||
"total_packets": struct.unpack_from("<I", head, 14)[0],
|
||
"payload": raw[20:20 + plen],
|
||
"raw": raw,
|
||
}
|
||
|
||
|
||
# --- Сборка сессии -------------------------------------------------------------
|
||
def assemble_session(packets):
|
||
if not packets:
|
||
raise SolveError("нет валидных пакетов CATD")
|
||
# Фильтр сессии: мода session_id; прочие сессии отбрасываются.
|
||
counts = Counter(p["session_id"] for p in packets)
|
||
session_id = counts.most_common(1)[0][0]
|
||
own = [p for p in packets if p["session_id"] == session_id]
|
||
dropped = len(packets) - len(own)
|
||
if dropped:
|
||
print(f"solver: отброшено {dropped} пакетов чужой сессии", file=sys.stderr)
|
||
totals = {p["total_packets"] for p in own}
|
||
if len(totals) != 1:
|
||
raise SolveError("несогласованные total_packets в сессии")
|
||
total = totals.pop()
|
||
if total < 3 or total > 100_000:
|
||
raise SolveError(f"недопустимый total_packets: {total}")
|
||
# Дубли: точные — удалить; конфликтующие (тот же sequence, другие байты) — отказ.
|
||
by_seq = {}
|
||
for p in own:
|
||
seq = p["sequence"]
|
||
if seq >= total:
|
||
raise SolveError(f"sequence {seq} вне total_packets {total}")
|
||
if seq in by_seq:
|
||
if by_seq[seq]["raw"] != p["raw"]:
|
||
raise SolveError(f"конфликтующий дубль sequence {seq}")
|
||
continue # точный дубль
|
||
by_seq[seq] = p
|
||
# Полнота.
|
||
missing = [s for s in range(total) if s not in by_seq]
|
||
if missing:
|
||
raise SolveError(f"неполная сессия: отсутствуют sequence {missing[:8]}"
|
||
f"{'...' if len(missing) > 8 else ''}")
|
||
ordered = [by_seq[s] for s in range(total)] # сортировка по (session_id, sequence)
|
||
if ordered[0]["type"] != TYPE_START:
|
||
raise SolveError("sequence 0 не START")
|
||
if ordered[-1]["type"] != TYPE_END:
|
||
raise SolveError("последний sequence не END")
|
||
for p in ordered[1:-1]:
|
||
if p["type"] != TYPE_DATA:
|
||
raise SolveError(f"sequence {p['sequence']} не DATA")
|
||
# Проверка END: первые 32 байта payload — SHA-256 от конкатенации
|
||
# payload DATA в порядке sequence; остаток — ASCII-хвост с правилами
|
||
# сборки (информативный, не проверяется).
|
||
stream = b"".join(p["payload"] for p in ordered[1:-1])
|
||
end_payload = ordered[-1]["payload"]
|
||
if len(end_payload) < 32:
|
||
raise SolveError("END payload короче 32 байт (нет SHA-256)")
|
||
if hashlib.sha256(stream).digest() != end_payload[:32]:
|
||
raise SolveError("END SHA-256 не сошёлся")
|
||
if end_payload[32:]:
|
||
print(f"solver: правила из END: "
|
||
f"{end_payload[32:].decode('ascii', 'replace')[:80]}...", file=sys.stderr)
|
||
return session_id, ordered[0]["payload"], stream
|
||
|
||
|
||
# --- Язык сцены ---------------------------------------------------------------
|
||
INT32_MIN, INT32_MAX = -2**31, 2**31 - 1
|
||
|
||
|
||
def parse_commands(b):
|
||
"""Разбор потока команд языка сцены (строгие длины). Команда — кортеж:
|
||
(OP_DEF_TEMPLATE, tid), (OP_END_DEF,), (OP_LINE, x1, y1, x2, y2),
|
||
(OP_MARK, id, x, y), (OP_GROUP, gid), (OP_END_GROUP,),
|
||
(OP_TRANSLATE, dx, dy), (OP_ROT90, k), (OP_REFLECT, axis),
|
||
(OP_INSTANCE, tid), (OP_DELETE, oid), (OP_PUBLISH, gid)."""
|
||
cmds = []
|
||
i = 0
|
||
|
||
def need(n, name):
|
||
if len(b) - i < n:
|
||
raise SolveError(f"усечённая команда {name}")
|
||
|
||
while i < len(b):
|
||
op = b[i]
|
||
i += 1
|
||
if op in (OP_END_DEF, OP_END_GROUP):
|
||
cmds.append((op,))
|
||
elif op in (OP_DEF_TEMPLATE, OP_GROUP, OP_INSTANCE, OP_DELETE, OP_PUBLISH):
|
||
need(2, "u16")
|
||
(ref,) = struct.unpack_from("<H", b, i)
|
||
i += 2
|
||
cmds.append((op, ref))
|
||
elif op == OP_ROT90:
|
||
need(1, "ROT90")
|
||
cmds.append((op, b[i]))
|
||
i += 1
|
||
elif op == OP_REFLECT:
|
||
need(1, "REFLECT")
|
||
cmds.append((op, b[i]))
|
||
i += 1
|
||
elif op == OP_TRANSLATE:
|
||
need(8, "TRANSLATE")
|
||
dx, dy = struct.unpack_from("<ii", b, i)
|
||
i += 8
|
||
cmds.append((op, dx, dy))
|
||
elif op == OP_MARK:
|
||
need(9, "MARK")
|
||
mid = b[i]
|
||
x, y = struct.unpack_from("<ii", b, i + 1)
|
||
i += 9
|
||
cmds.append((op, mid, x, y))
|
||
elif op == OP_LINE:
|
||
need(16, "LINE")
|
||
x1, y1, x2, y2 = struct.unpack_from("<iiii", b, i)
|
||
i += 16
|
||
cmds.append((op, x1, y1, x2, y2))
|
||
else:
|
||
raise SolveError(f"неизвестная команда сцены {op}")
|
||
return cmds
|
||
|
||
|
||
def _apply_transform(t, x, y):
|
||
op = t[0]
|
||
if op == OP_TRANSLATE:
|
||
return x + t[1], y + t[2]
|
||
if op == OP_ROT90:
|
||
for _ in range(t[1] % 4):
|
||
x, y = -y, x
|
||
return x, y
|
||
if op == OP_REFLECT:
|
||
if t[1] == 0:
|
||
return -x, y
|
||
if t[1] == 1:
|
||
return x, -y
|
||
raise SolveError(f"REFLECT с недопустимой осью {t[1]}")
|
||
raise SolveError(f"команда {op} не преобразование")
|
||
|
||
|
||
def _transform_point(chain, x, y):
|
||
"""Цепочка преобразований: сначала список своей группы (chain[0], в
|
||
порядке появления), затем родительской, и так до корня."""
|
||
for ts in chain:
|
||
for t in ts:
|
||
x, y = _apply_transform(t, x, y)
|
||
if not (INT32_MIN <= x <= INT32_MAX and INT32_MIN <= y <= INT32_MAX):
|
||
raise SolveError(f"переполнение координат сцены ({x}, {y})")
|
||
return x, y
|
||
|
||
|
||
def execute_scene(cmds):
|
||
"""Исполнение потока команд языка сцены. Возвращает (lines, marks)
|
||
итоговой ОПУБЛИКОВАННОЙ сцены в мировых координатах:
|
||
lines = [(x1, y1, x2, y2)], marks = [(id, x, y)].
|
||
|
||
Семантика (спека в START payload): шаблоны содержат LINE/MARK в
|
||
локальных координатах; INSTANCE присваивает object id в порядке
|
||
чтения; DELETE удаляет объект в момент чтения команды; преобразование
|
||
группы — композиция её TRANSLATE/ROT90/REFLECT в порядке появления,
|
||
применяется ко всему телу; мировая точка — преобразование своей
|
||
группы, затем родительской, до корня; PUBLISH ровно один, считается
|
||
только опубликованная сцена."""
|
||
templates = {}
|
||
groups = {} # gid -> {"transforms": [...], "items": [...]}
|
||
objects = {} # oid -> группа-владелец
|
||
publish_gid = None
|
||
next_oid = 0
|
||
stack = [] # пустой = верхний уровень
|
||
in_def = None
|
||
def_body = []
|
||
|
||
for pos, c in enumerate(cmds):
|
||
op = c[0]
|
||
if in_def is not None:
|
||
if op == OP_END_DEF:
|
||
templates[in_def] = def_body
|
||
in_def, def_body = None, []
|
||
elif op in (OP_LINE, OP_MARK):
|
||
def_body.append(c)
|
||
else:
|
||
raise SolveError(f"команда {op} в теле DEF_TEMPLATE (поз. {pos})")
|
||
continue
|
||
if op == OP_DEF_TEMPLATE:
|
||
if stack:
|
||
raise SolveError("DEF_TEMPLATE внутри группы")
|
||
if c[1] in templates:
|
||
raise SolveError(f"повторный DEF_TEMPLATE tid={c[1]}")
|
||
in_def = c[1]
|
||
elif op == OP_END_DEF:
|
||
raise SolveError("END_DEF вне DEF_TEMPLATE")
|
||
elif op == OP_GROUP:
|
||
if c[1] in groups:
|
||
raise SolveError(f"повторный GROUP gid={c[1]}")
|
||
g = {"gid": c[1], "transforms": [], "items": []}
|
||
groups[c[1]] = g
|
||
if stack:
|
||
stack[-1]["items"].append(("group", g))
|
||
stack.append(g)
|
||
elif op == OP_END_GROUP:
|
||
if not stack:
|
||
raise SolveError("END_GROUP вне группы")
|
||
stack.pop()
|
||
elif op in (OP_TRANSLATE, OP_ROT90, OP_REFLECT):
|
||
if not stack:
|
||
raise SolveError("преобразование вне группы")
|
||
if op == OP_REFLECT and c[1] > 1:
|
||
raise SolveError(f"REFLECT с недопустимой осью {c[1]}")
|
||
stack[-1]["transforms"].append(c)
|
||
elif op == OP_INSTANCE:
|
||
if not stack:
|
||
raise SolveError("INSTANCE вне группы")
|
||
if c[1] not in templates:
|
||
raise SolveError(f"INSTANCE неизвестного шаблона tid={c[1]}")
|
||
next_oid += 1
|
||
if next_oid > 0xFFFF:
|
||
raise SolveError("переполнение object id")
|
||
stack[-1]["items"].append(("inst", next_oid, c[1]))
|
||
objects[next_oid] = stack[-1]
|
||
elif op in (OP_LINE, OP_MARK):
|
||
if not stack:
|
||
raise SolveError("LINE/MARK вне группы/шаблона")
|
||
stack[-1]["items"].append(("cmd", c))
|
||
elif op == OP_DELETE:
|
||
if stack:
|
||
raise SolveError("DELETE внутри группы")
|
||
owner = objects.get(c[1])
|
||
if owner is None:
|
||
raise SolveError(f"DELETE неизвестного object id {c[1]}")
|
||
for k, it in enumerate(owner["items"]):
|
||
if it[0] == "inst" and it[1] == c[1]:
|
||
del owner["items"][k]
|
||
break
|
||
else:
|
||
raise SolveError(f"object id {c[1]} уже удалён")
|
||
del objects[c[1]]
|
||
elif op == OP_PUBLISH:
|
||
if stack:
|
||
raise SolveError("PUBLISH внутри группы")
|
||
if publish_gid is not None:
|
||
raise SolveError("повторный PUBLISH")
|
||
if c[1] not in groups:
|
||
raise SolveError(f"PUBLISH неизвестной группы gid={c[1]}")
|
||
publish_gid = c[1]
|
||
else:
|
||
raise SolveError(f"неизвестная команда сцены {op} (поз. {pos})")
|
||
if in_def is not None:
|
||
raise SolveError(f"DEF_TEMPLATE tid={in_def} не закрыт")
|
||
if stack:
|
||
raise SolveError(f"GROUP gid={stack[-1]['gid']} не закрыта")
|
||
if publish_gid is None:
|
||
raise SolveError("нет PUBLISH — сцена не опубликована")
|
||
|
||
lines, marks = [], []
|
||
|
||
def emit(chain, c):
|
||
if c[0] == OP_LINE:
|
||
x1, y1 = _transform_point(chain, c[1], c[2])
|
||
x2, y2 = _transform_point(chain, c[3], c[4])
|
||
lines.append((x1, y1, x2, y2))
|
||
elif c[0] == OP_MARK:
|
||
x, y = _transform_point(chain, c[2], c[3])
|
||
marks.append((c[1], x, y))
|
||
|
||
def walk(g, chain):
|
||
chain = [g["transforms"]] + chain
|
||
for it in g["items"]:
|
||
if it[0] == "group":
|
||
walk(it[1], chain)
|
||
elif it[0] == "inst":
|
||
for tc in templates[it[2]]:
|
||
emit(chain, tc)
|
||
else:
|
||
emit(chain, it[1])
|
||
|
||
walk(groups[publish_gid], [])
|
||
return lines, marks
|
||
|
||
|
||
def render_scene(lines, marks, path):
|
||
"""Отрисовка опубликованной сцены: LINE в мировых координатах (Pillow)."""
|
||
if not lines:
|
||
raise SolveError("нечего рисовать")
|
||
xs = [v for ln in lines for v in (ln[0], ln[2])]
|
||
ys = [v for ln in lines for v in (ln[1], ln[3])]
|
||
x0, x1 = min(xs) / 1000.0, max(xs) / 1000.0
|
||
y0, y1 = min(ys) / 1000.0, max(ys) / 1000.0
|
||
margin = 20.0
|
||
scale = 800.0 / max(x1 - x0 + 2 * margin, y1 - y0 + 2 * margin, 1.0)
|
||
w = int((x1 - x0 + 2 * margin) * scale)
|
||
h = int((y1 - y0 + 2 * margin) * scale)
|
||
|
||
def tr(x, y):
|
||
return ((x / 1000.0 - x0 + margin) * scale,
|
||
(y / 1000.0 - y0 + margin) * scale)
|
||
|
||
img = Image.new("RGB", (w, h), (250, 250, 248))
|
||
dr = ImageDraw.Draw(img)
|
||
width_px = max(2, int(scale * 3))
|
||
for lx1, ly1, lx2, ly2 in lines:
|
||
dr.line([tr(lx1, ly1), tr(lx2, ly2)], fill=(20, 20, 20), width=width_px)
|
||
for mid, mx, my in marks:
|
||
cx, cy = tr(mx, my)
|
||
r = 6
|
||
col = {1: (200, 30, 30), 2: (30, 30, 200), 3: (30, 160, 60)}.get(mid, (0, 0, 0))
|
||
dr.ellipse([cx - r, cy - r, cx + r, cy + r], outline=col, width=3)
|
||
img.save(path)
|
||
|
||
|
||
# --- Публичная функция флага (SPECS.md 5.4; копия логики noseflag) ----------------
|
||
def parse_thousandths(s):
|
||
orig = s
|
||
if s == "":
|
||
raise SolveError(f'E_FORMAT: invalid decimal "{orig}"')
|
||
neg = False
|
||
if s[0] in "+-":
|
||
neg = s[0] == "-"
|
||
s = s[1:]
|
||
if "." in s:
|
||
int_part, frac_part = s.split(".", 1)
|
||
else:
|
||
int_part, frac_part = s, None
|
||
if int_part == "" or (frac_part is not None and frac_part == ""):
|
||
raise SolveError(f'E_FORMAT: invalid decimal "{orig}"')
|
||
if not int_part.isascii() or not int_part.isdigit():
|
||
raise SolveError(f'E_FORMAT: invalid decimal "{orig}"')
|
||
if frac_part is not None and (not frac_part.isascii() or not frac_part.isdigit()):
|
||
raise SolveError(f'E_FORMAT: invalid decimal "{orig}"')
|
||
if frac_part is not None and len(frac_part) > 3:
|
||
raise SolveError(f'E_PRECISION: more than 3 fractional digits in "{orig}"')
|
||
frac = (frac_part or "").ljust(3, "0")
|
||
if len(int_part) > 7:
|
||
raise SolveError(f'E_RANGE: out of range [-1000.000, +1000.000]: "{orig}"')
|
||
v = int(int_part) * 1000 + int(frac)
|
||
if neg:
|
||
v = -v
|
||
if v < -1000000 or v > 1000000:
|
||
raise SolveError(f'E_RANGE: out of range [-1000.000, +1000.000]: "{orig}"')
|
||
return v
|
||
|
||
|
||
def flag_from_thousandths(x, y):
|
||
u = x + 1000000
|
||
v = y + 1000000
|
||
z = u * FLAG_MODULUS + v
|
||
return f"caplag{{nose_{z}}}"
|
||
|
||
|
||
def thousandths_to_decimal(v):
|
||
sign = "-" if v < 0 else ""
|
||
a = abs(v)
|
||
return f"{sign}{a // 1000}.{a % 1000:03d}"
|
||
|
||
|
||
# --- main -------------------------------------------------------------------------
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(description="Решатель задачи 5 (CATD)")
|
||
ap.add_argument("png", help="участнический стартовый PNG")
|
||
ap.add_argument("--render", help="куда сохранить отрисовку опубликованной сцены")
|
||
args = ap.parse_args(argv)
|
||
|
||
try:
|
||
with open(args.png, "rb") as f:
|
||
start_png = f.read()
|
||
pcapng = extract_pcapng(start_png)
|
||
print(f"solver: PCAPNG {len(pcapng)} байт", file=sys.stderr)
|
||
interfaces, packets = parse_pcapng(pcapng)
|
||
print(f"solver: пакетов в захвате: {len(packets)}", file=sys.stderr)
|
||
bodies = extract_http_png_bodies(interfaces, packets)
|
||
print(f"solver: HTTP PNG-тел: {len(bodies)}", file=sys.stderr)
|
||
if not bodies:
|
||
raise SolveError("в захвате не найдено POST /upload с PNG")
|
||
decoded = []
|
||
for i, body in enumerate(bodies):
|
||
try:
|
||
decoded.append(decode_packet(body))
|
||
except SolveError as e:
|
||
raise SolveError(f"PNG #{i}: {e}")
|
||
session_id, start_payload, stream = assemble_session(decoded)
|
||
print(f"solver: сессия {session_id}, START: "
|
||
f"{start_payload.decode('ascii', 'replace')[:60]}...", file=sys.stderr)
|
||
cmds = parse_commands(stream)
|
||
print(f"solver: команд языка сцены: {len(cmds)}", file=sys.stderr)
|
||
lines, marks = execute_scene(cmds)
|
||
print(f"solver: опубликованная сцена: линий {len(lines)}, "
|
||
f"маркеров {len(marks)}", file=sys.stderr)
|
||
noses = [m for m in marks if m[0] == 1]
|
||
if not noses:
|
||
raise SolveError("MARK id=1 (нос) не найден в опубликованной сцене")
|
||
if len(noses) > 1:
|
||
raise SolveError("MARK id=1 не единственный в опубликованной сцене")
|
||
_, nx, ny = noses[0]
|
||
xs, ys = thousandths_to_decimal(nx), thousandths_to_decimal(ny)
|
||
print(f"solver: нос (мировые) X={xs} Y={ys}", file=sys.stderr)
|
||
flag = flag_from_thousandths(parse_thousandths(xs), parse_thousandths(ys))
|
||
if args.render:
|
||
render_scene(lines, marks, args.render)
|
||
print(f"solver: опубликованная сцена сохранена в {args.render}",
|
||
file=sys.stderr)
|
||
print(flag)
|
||
return 0
|
||
except SolveError as e:
|
||
print(f"solver: ошибка: {e}", file=sys.stderr)
|
||
return 1
|
||
except (zlib.error, struct.error) as e:
|
||
print(f"solver: ошибка формата: {e}", file=sys.stderr)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|