Initial commit: MUD client v0.2

This commit is contained in:
truetype committed 2026-07-07 15:38:28 +00:00
commit 0030fd7d82
26 files changed
+4421

No files matched your search

+65
View File
@@ -0,0 +1,65 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# Virtual environments
.venv/
venv/
env/
ENV/
env.bak/
venv.bak/
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.pytest_cache/
# mypy / type checkers
.mypy_cache/
.dmypy.json
dmypy.json
.pyre/
# IDEs / editors
.idea/
.vscode/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
# Jupyter
.ipynb_checkpoints/
+210
View File
@@ -0,0 +1,210 @@
# progetto2
`progetto2` ospita un **client MUD CLI** pensato per collegarsi a un MUD
LP‑MUD basato sul driver **FluffOS** (per esempio **Dead Souls**, libreria
3.8.6). Il client parla telnet con estensioni standard: `ECHO`, `NAWS`,
`TTYPE`, e rifiuta (`WONT`) tutte le altre (`GMCP`, `MSDP`, `MCCP`, `MXP`
per ora). Tiene traccia delle stanze visitate e le renderizza come **mappa
ASCII** on-demand.
## Requisiti
- Python ≥ 3.10
- Una libreria Python: [`prompt_toolkit`](https://pypi.org/project/prompt-toolkit/) ≥ 3.0
## Setup
```bash
cd progetto2
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
```
Verrà installato il comando `mudclient`.
## Uso
```bash
# Default: si collega a localhost:6666 con un finestra iniziale 120×40
mudclient
# Espliciti
mudclient --host localhost --port 6666 --cols 120 --rows 40
# Auto-login: passa il nome; il client lo invia quando il server chiede
# "What name do you wish?". Se serve la password, il client chiede
# /password PWD oppure la puoi passare con --password (o env MUDCLIENT_PASSWORD).
mudclient --user zork --password hunter2
# Log di sessione: ogni evento (testo, GMCP, stanze, comandi locali, linee inviate)
# viene serializzato in append‑only, flushed dopo ogni write — adatto a `tail -f`.
mudclient --log-json ~/mud-session.jsonl --log-md ~/mud-session.md
```
> I file `--log-json` / `--log-md` vengono **truncati** all'apertura:
> ogni nuova sessione ricomincia da zero. Per ruotare manualmente
> prima della sessione:
> ```bash
> mv ~/mud-session.jsonl{,.prev} && mudclient --log-json ~/mud-session.jsonl
> ```
Schema eventi (un oggetto JSON per riga):
| `kind` | Scopo | Note |
| --- | --- | --- |
| `session_start` | apertura sessione | `host`, `port`, `user` |
| `session_end` | chiusura normale | `reason` |
| `disconnect` | chiusura socket | `reason` |
| `text` | chunk ANSI ricevuto dal server | `raw` preservato |
| `room_enter` | mapper ha registrato una stanza | `id`, `name`, `exits` |
| `gmcp` | frame GMCP accettato | `module`, `payload` |
| `vitals` | snapshot hp/sp/mv | solo i campi presenti |
| `line_sent` | riga inoltrata al MUD | password redatte (`***`) |
| `local_cmd` | comando `/...` intercettato | password/segreti redatti |
| `login` | prompt/risposta auto-login | `value` redatto di default; passare `redact=False` per i nomi account |
Una volta connesso:
| Tasto | Effetto |
| --- | --- |
| Testo qualsiasi | inviato al MUD terminato da `\r\n` |
| `/map` | disegna la mappa ASCII delle stanze visitate (multi-source BFS) |
| `/where` | mostra la stanza corrente e le uscite |
| `/look` | invia `look` al MUD |
| `/vitals` | mostra hp/sp/mv (via GMCP `Char.Vitals`) |
| `/stats` | locali scoperti, GMCP stato, vitali |
| `/password PWD` | invia una password al MUD |
| `/clear` | pulisce lo schermo |
| `/help` (`/?`) | mostra l'aiuto |
| `/quit` (`/q`, `Ctrl‑D`) | esce |
Il prompt è sempre del tipo:
```
[NomeStanza]> _
```
con il nome della stanza corrente in cyan brillante. In basso c'è una **status bar** che mostra:
```
[stanza: Outer Courtyard] [GMCP: on] HP 100/100 SP 50/50 MV 80/80
```
Quando il server invia `Char.Vitals` via GMCP, hp/sp/mv compaiono automaticamente.
## Layout dei sorgenti
```
progetto2/
├── pyproject.toml # metadata + dipendenze + entry point `mudclient`
├── README.md # questo file
├── .gitignore
├── docs/ # documentazione completa (vedi sotto)
│ ├── README.md # indice della cartella docs
│ ├── architecture.md # analisi architetturale (data flow + boundary)
│ ├── modules.md # descrizione modulo‑per‑modulo
│ ├── protocol.md # telnet, GMCP, ANSI, euristiche Dead Souls
│ ├── usage.md # flag CLI, esempi, auto‑login, comandi slash
│ ├── events.md # schema eventi del logger
│ ├── development.md # setup dev, pytest, ruff, come estendere
│ └── gitea.md # flusso di pubblicazione sul Gitea + sicurezza token
├── src/progetto2/ # package
│ ├── __init__.py
│ ├── __main__.py # supporta `python -m progetto2`
│ ├── cli.py # entry point + UI a riga di comando
│ ├── telnet.py # state machine IAC + asyncio
│ ├── deadsouls.py # parser di stanze Dead Souls / LP‑MUD
│ ├── gmcp.py # parser GMCP (Char.Vitals / Room.Info)
│ ├── mapper.py # grafo delle stanze + render ASCII
│ ├── logger.py # EventLogger (JSONL + Markdown sinks)
│ └── ansi.py # stripper ANSI per il parser
└── tests/ # unit test (pytest)
├── test_telnet.py
├── test_ansi.py
├── test_deadsouls.py
├── test_mapper.py
└── test_logger.py
```
## 📚 Documentazione completa
La documentazione vive nella cartella [`docs/`](./docs/README.md) e
copre:
- **[`docs/architecture.md`](./docs/architecture.md)** — analisi
architetturale completa: dipendenze tra moduli, data flow di un
frame dal server, boundary concerns, refactor raccomandati.
- **[`docs/modules.md`](./docs/modules.md)** — cosa fa ogni modulo,
API pubblica, responsabilità, tolleranze.
- **[`docs/protocol.md`](./docs/protocol.md)** — IAC state machine,
GMCP wire format, euristiche Dead Souls, ANSI handling.
- **[`docs/usage.md`](./docs/usage.md)** — riferimento flag CLI, esempi
pratici, flusso auto‑login, comandi slash, rotazione log.
- **[`docs/events.md`](./docs/events.md)** — schema completo degli
eventi del logger (JSONL + Markdown), regole di redazione, esempi
`jq`.
- **[`docs/development.md`](./docs/development.md)** — workflow
sviluppatore: setup, test, lint, come aggiungere moduli, debugging.
- **[`docs/gitea.md`](./docs/gitea.md)** — flusso di pubblicazione sul
server Gitea `192.168.0.183:8085` (utente `truetype`) **+ regole di
sicurezza del token**.
## Note sul protocollo
Il client negozia:
- `ECHO`: invia `WILL ECHO` in risposta a `DO ECHO`. Il server non eseguirà
più echo remoto, mentre l'eco locale è gestita da `prompt_toolkit`.
- `NAWS`: se attivo, invia la dimensione della finestra iniziale.
- `TTYPE`: risponde con `XTERM-256COLOR` quando il server lo richiede.
- `GMCP`: viene **offerto proattivamente** (`WILL` al connect). Dead Souls
lo accetta e invia pacchetti `Char.Vitals`, `Room.Info`, `Char.Login`,
ecc. Quando una stanza arriva via `Room.Info`, si usa il `num`
(vnum) del server come id stabile della stanza.
Le righe di stato del parser considerano `IAC GA` (`249`) e `IAC EOR`
(`239`) come delimitatori di "record/prompt", così da poter chiudere un
blocco di stanza anche quando il prompt testuale non è immediatamente
riconoscibile.
La rilevazione automatica della geometria del mondo si basa sul **movimento
inviato dall'utente** + **descrizione della stanza** osservata subito
dopo: quando digiti `n`, il client informa il mapper; quando la risposta
del MUD arriva, `Mapper.visit()` collega automaticamente
`corrente → nuova` lungo la direzione digitata.
Il render della mappa è un **multi-source BFS** dal current room, con
fallback di **posizionamento orfano** per le stanze visitate in sequenza
indietro (es. `n→s`) e stack nell'angolo in basso a destra per stanze
completamente isolate.
## Limitazioni note (v0.2)
- **MSDP / MCCP / MCP / MXP** non sono ancora abilitati: una sessione che
li richiede ottiene `WONT` dal nostro lato. Dead Souls usa GMCP, che è
ben coperto; altri MUD basati su MSDP possono essere aggiunti seguendo
lo schema di `gmcp.py`.
- La detection delle stanze **senza GMCP** resta **euristica** sul testo.
- **Login automatico**: il client riconosce il prompt `What name do you
wish?` / `Password:` e invia automaticamente solo la prima volta.
Per la password senza `--password` viene mostrato un suggerimento
`usa /password <pwd>` (niente prompt annidato, troppo fragile).
- Gli `u`/`d` (verticali) sono rappresentati nella mappa ma non
influenzano le coordinate 2D (offset `(0,0)`).
- La status bar si aggiorna ad ogni rilettura della riga del prompt;
per refresh continui in idle si può aggiungere `refresh_interval`
al `PromptSession` come estensione futura.
## Sviluppo
```bash
pip install -e ".[dev]"
pytest # gira tutta la suite
ruff check src tests # lint
```
## Licenza
MIT per il codice del client. Il server **Dead Souls** non è GPL — assicurati
di rispettarne la licenza prima di redistribuirlo.
+41
View File
@@ -0,0 +1,41 @@
# `docs/` — indice della documentazione di `progetto2`
Documentazione completa del client MUD CLI. La lingua è l'italiano per
coerenza con il `README.md` principale. I blocchi di codice e i nomi
di simboli rimangono in inglese.
| Sezione | Cosa trovi |
| --- | --- |
| [`architecture.md`](./architecture.md) | Analisi architetturale: dipendenze tra moduli, data flow, boundary, accoppiamenti noti. **Leggi prima questo.** |
| [`modules.md`](./modules.md) | Descrizione modulo‑per‑modulo: scopo, API pubblica, dipendenze interne, slot/responsabilità. |
| [`protocol.md`](./protocol.md) | Telnet (IAC state machine), GMCP wire format, euristiche Dead Souls, ANSI handling. |
| [`usage.md`](./usage.md) | Tutti i flag della CLI + esempi + flusso di auto‑login + comandi locali `/...`. |
| [`events.md`](./events.md) | Schema completo degli eventi emessi dal `EventLogger` (JSONL + Markdown). |
| [`development.md`](./development.md) | Setup dev, `pytest`, `ruff`, come aggiungere un modulo, debugging. |
| [`gitea.md`](./gitea.md) | Flusso di pubblicazione sul server Gitea `192.168.0.183:8085` (utente `truetype`) + **regole di sicurezza del token**. |
## Biglietto da visita
```
┌──────────────┐
│ cli.py │ ← orchestratore async (prompt / dispatcher)
└──────┬───────┘
│
┌───────────┼─────────────┐
│ │ │
┌─▼────┐ ┌────▼─────┐ ┌─────▼──────┐
│logger│ │ telnet │ │ mapper │
└──────┘ └─────┬────┘ └─────┬──────┘
│ │
│ GMCP ▲ via deadsouls (testo)
┌────────▼────────┐ │
│ gmcp │ │
└─────────────────┘ │
│
┌───────▼────────┐
│ deadsouls │
└────────────────┘
+ ansi (foglia pura)
```
Per la versione completa e discorsa vedi [`architecture.md`](./architecture.md).
+167
View File
@@ -0,0 +1,167 @@
# Architettura di `progetto2`
> Versione: v0.2.x — client MUD CLI asincrono in Python ≥ 3.10 basato su
> `prompt_toolkit`. Niente dipendenze pesanti: solo la libreria del
> prompt + `pytest/ruff` per la dev.
## Visione d'insieme
Il client è organizzato in **cinque pilastri** indipendenti, composti
da `cli.py` che fa da orchestratore asincrono:
| Pilastro | Modulo | Ruolo |
| --- | --- | --- |
| **Trasporto** | `telnet.py` | State machine IAC + asyncio TCP + negoziazione opzioni (ECHO/NAWS/TTYPE/GMCP a livello di byte‑option; non parsa i payload). Nessuna conoscenza del MUD. |
| **Euristiche di testo** | `deadsouls.py` | Parsing delle stanze Dead Souls / LP‑MUD: canonicalizzazione direzioni, rilevazione blocchi stanza, regex uscite. |
| **Payload semantici** | `gmcp.py` | Decoder GMCP: validazione *di forma* del module name + estrazione tollerante `Char.Vitals` / `Room.Info`. |
| **Grafo & layout** | `mapper.py` | Database stanze (id‑stable), BFS multi‑source per coordinate, render ASCII del grafo. |
| **Osservabilità** | `logger.py` | Sink JSONL append‑only + Markdown streamed; redazione password; lock difensivo. |
| **UI** | `cli.py` | Orchestrazione: prompt async, status bar, dispatcher telnet, auto‑login, comandi slash, terminazione pulita. |
A parte `ansi.py` (foglia pura: solo regex + `re.sub`), tutti gli altri
moduli sono foglie funzionali tranne `cli.py`.
## Grafo delle dipendenze
```
┌─────────────────────┐
│ cli.py │ (orchestratore)
└────────┬────────────┘
│ importa
┌───────────────┬─────┴──────┬───────────────┬──────────────┐
│ │ │ │ │
┌──▼─────┐ ┌─────▼─────┐ ┌───▼──────┐ ┌───────▼──────┐ ┌─────▼──────┐
│ logger │ │ telnet │ │ gmcp │ │ mapper │ │ deadsouls │
└────────┘ └─────┬─────┘ └─────┬────┘ └───────┬──────┘ └─────┬──────┘
│ │ │ │
│ SubnegEvent │ │ import │
│ (OPT_GMCP) │ │ EXIT_ALIASES │
│ │ EXIT_ALIASES│ + offsets │
│ ├──┐ ├──────────────┘
│ │ └───────────┘
│ │
│ ascii parser (IACParser)
└─ nessuna dipendenza da altri moduli MUD
│
┌───▼──────┐
│ ansi.py │ (foglia pura)
└──────────┘
```
**Regola di import**: `gmcp.py` e `mapper.py` **non** dovrebbero
importare da `cli.py` (e non lo fanno). `telnet.py` non sa nulla di
GMCP: emette solo `SubnegEvent(OPT_GMCP, data)` e lascia che `cli.py`
lo smisti alla callback di `_handle_gmcp`.
## Data flow di un frame in ingresso
```
server TCP
│
│ bytes grezzi
▼
┌──────────────────────────────────────────┐
│ TelnetConnection.read() │ chunk = 4096 byte
└──────┬───────────────────────────────────┘
│
▼
┌──────────────────────────────────────────┐
│ IACParser.feed(chunk) │ state machine
│ • S.NORMAL → accumula testo │
│ • S.IAC → DO/DONT/WILL/WONT/SB │
│ • S.SB_* → accumulator dati subneg │
│ • EOR/GA → flush + PromptEvent │
└──────┬───────────────────────────────────┘
│ list[TelnetEvent]
▼
┌──────────────────────────────────────────┐
│ conn.read_loop(handler) │ smistamento (TUTTI gli eventi
│ • tutti gli eventi passano anche per │ passano anche per `handler`):
│ l'handler; Negotiate/Subneg vengono │ • NegotiateEvent → _on_control
│ PRIMA processati da _on_control, │ (risposta WILL/WONT/DONT)
│ POI inoltrati all'utente │ • SubnegEvent → handler → cli
│ │ → _handle_gmcp (se OPT_GMCP)
│ │ • TextEvent → print + log
│ │ • PromptEvent → marker (no‑op) │
└──────┬───────────────────────────────────┘
│ TextEvent | SubnegEvent
▼
┌──────────────────────────────────────────┐
│ _on_telnet_event (cli.py) │
│ • TextEvent → MAPPER (via deadsouls) │
│ • TextEvent → auto‑login (regex) │
│ • SubnegEvent(GMCP) → gmcp.parse + │
│ state.update │
└──────────────────────────────────────────┘
```
Nessun modulo a valle dell'IAC si preoccupa di IAC: è un boundary
pulito. `gmcp.py` riceve **solo** i `bytes` della subneg, non il
flusso completo. `deadsouls.RollingRoomParser` riceve **solo testo
plain** (ANSI già strippato da `cli.py`).
## Boundary concerns (punti deboli noti)
1. **Canonicalizzazione direzioni duplicata.**
`deadsouls.EXIT_ALIASES` è la sorgente canonica. Tuttavia
`mapper.py`, `gmcp.py` e `cli.py` (via `_handle_gmcp:exits_list`)
chiamano tutti `EXIT_ALIASES.get(d, d)` su ogni uscita. **Rischio
reale**: se domani aggiungi un alias in `EXIT_ALIASES` e dimentichi
di aggiornare la logica del consumer, ottieni nomi non canonicali.
- *Mitigazione breve*: aggiungere `deadsouls.normalize_exits(list)`
e farla usare ovunque.
- *Mitigazione strutturale*: promuovere `EXIT_ALIASES` +
`DIRECTION_OFFSETS` in un modulo `navigation.py` da cui
`deadsouls` *re‑importi* per retro‑compatibilità.
2. **Doppia fonte "camera corrente".**
`mapper.Mapper._current` è la verità. Ma `cli.ClientState` espone
`room_name` come property che rilancia su `mapper.current_room`. Non
c'è cache in `cli`, quindi sono sincronizzati, ma l'oggetto
`ClientState` ha più campi di quanti `_run()` ne inizializzi alcuni
(`auto_password_prompted`, `vitals`) — è verboso ma non buggy.
3. **Re‑entrata del logger in `__init__`.**
`EventLogger.__init__` apre due file in scrittura + scrive l'header
MD; un errore sulla seconda operazione lascia il primo file aperto
fino al GC. È mitigato da `try/except OSError` + `_close_files()`,
ma l'eccezione rilanciata può lasciare un file orfano sul disco se
il kernel non chiude immediatamente. In pratica: il `finally` di
`_run` chiama `state.logger.close()` e quindi GC non è mai
l'unico meccanismo.
4. **Validazione GMCP solo di *shape*, non di *semantica*.**
La regex `_MODULE_NAME_RE = ^[A-Za-z][A-Za-z0-9_.]*$` è ancorata
e rifiuta correttamente nomi malformati (`.bad`, `1Room`,
`Room..Info`). Tuttavia non valida la *semantica* del payload:
un `Room.Info` con `exits` come stringa invece di `dict`/`list`
produce silenziosamente una lista uscite vuota. Documentato ma non
ancora controllato.
5. **`refresh_interval=0.5` su `PromptSession`.**
Anche in idle ridisegna la toolbar due volte al secondo. Costo CPU
trascurabile su laptop moderne ma visibile su terminali via SSH su
linee lente. Accettabile per v0.2; documentato come noto.
## Estensioni future raccomandate
| Estensione | Impatto | Dove intervenire |
| --- | --- | --- |
| `directions.py` (alias + offset + normalize) | alto (boundary) | spostare `EXIT_ALIASES`/`DIRECTION_OFFSETS` da `deadsouls.py`, re‑importare per back‑compat |
| MSDP / MCCP / MXP | medio | aggiungere costanti + handling in `telnet._negotiate`, modulo parser parallelo a `gmcp.py` |
| Persistenza stanze su disco | medio | `mapper.Mapper.to_json()` / `from_json()`, hook in `_run`/`_maybe_autologin` fine sessione |
| `EventLogger` su sink syslog/journald | basso | aggiungere `journald_path` a `__init__` + dispatcher |
| Multi‑sessione (multi `Mapper` + tab) | alto | refactoring `ClientState` in `Session`, lista in `cli` |
| mappa HTML/SVG | medio | nuovo `mapper.render_svg()` + renderer HTML minimale |
## Decisioni di design (perché così)
| Decisione | Alternativa scartata | Motivo |
| --- | --- | --- |
| `telnet.py` non importa `gmcp.py` | metodo `parse_gmcp(self, ev)` | injection di dipendenza esplicita + testing più semplice |
| Mappa lazy: coordinate solo a `render()` | coordinate in tempo reale | permette backtracking senza snapshot, layout deterministico al render |
| Logger append‑only senza buffering | buffer batch + flush periodico | `tail -f` immediato è OK per volumi MUD (kB/s non MB/s) |
| Lock difensivo nel logger | single‑thread assumption | costo trascurabile, protegge da futuri worker thread |
| Auto‑login regex multi‑linea | stateful prompt matcher | regex robusta + documentata + testabile |
| NAWS solo iniziale (no SIGWINCH handler) | re‑invio su cambio finestra | première itération: il client non intercetta segnali; documentato come gap |
| `EventLogger` come sink opt‑in (no default) | always‑on log | sovraccarico inutile su sessioni interattive |
+174
View File
@@ -0,0 +1,174 @@
# Sviluppo
Setup locale, pytest, ruff, e come estendere il client.
---
## Setup
```bash
git clone https://192.168.0.183:8085/truetype/progetto2.git
cd progetto2
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
```
Lo sviluppo quotidiano è Python puro; nessun build system esterno
(setuptools è trasparente via `pyproject.toml`).
---
## Test
```bash
pytest # 66+ test (telnet + ansi + deadsouls + mapper + logger)
pytest tests/test_logger.py -v # solo il logger (più verboso)
pytest -k "test_rolling_parser" # filtrato per nome
pytest --tb=short # traceback corti
```
I test usano `pytest.ini_options = ["-q"]` di default in `pyproject.toml`.
### Cosa è coperto
| File | Copre |
| --- | --- |
| `test_ansi.py` | stripper ANSI + `visible_len` |
| `test_telnet.py` | IACParser (state machine completa) + TelnetConnection (mock asyncio) |
| `test_deadsouls.py` | `parse_exits` + `RollingRoomParser` scenario multi‑chunk |
| `test_mapper.py` | `Mapper.visit`, link, layout BFS, orphan placement, render |
| `test_logger.py` | EventLogger round‑trip JSONL, MD rendering per ogni `kind`, redazione password, idempotenza `close()`, lone‑ESC safety |
### Cosa *non* è coperto
- `cli.py` end‑to‑end (richiede telnet server; il vero test è una
smoke run contro il docker locale di Dead Souls).
- `mapper._render_vertical_row` (casi degeneri con celle non
continue; non aggiunti perché non osservati in pratica).
- Re‑entrata del logger in un worker thread (lock difensivo è
documentato ma non testato).
---
## Lint
```bash
ruff check src tests
ruff format src tests
```
Config in `pyproject.toml`:
```toml
[tool.ruff]
line-length = 100
target-version = "py310"
```
(Aggiungi regole strict `select = ["E", "F", "I", "B", "UP"]` se
vuoi enforcement.)
---
## Smoke test end‑to‑end
Dead Souls docker su `localhost:6666`:
```bash
# 2 secondi, log JSONL + MD
LOGJSON=$(mktemp /tmp/mud-json.XXXXXX.jsonl)
LOGMD=$(mktemp /tmp/mud-md.XXXXXX.md)
timeout 2 python -m progetto2 --host localhost --port 6666 \
--log-json "$LOGJSON" --log-md "$LOGMD" </dev/null >/dev/null 2>&1 || true
wc -l "$LOGJSON" "$LOGMD"
head -3 "$LOGJSON"
head -8 "$LOGMD"
```
Aspettative: almeno 2‑3 righe JSONL (session_start, text, session_end)
e un header MD con `host/port/user`.
---
## Come aggiungere un modulo
Esempio: nuovo modulo `directions.py` che assorbe `EXIT_ALIASES` da
`deadsouls.py`. Procedura:
1. Crea `src/progetto2/directions.py` con le costanti e `normalize_exits()`.
2. Aggiorna `deadsouls.py` → re‑import da `directions` per back‑compat.
3. Aggiorna `gmcp.py`, `mapper.py`, `cli.py` → import da `directions`.
4. Scrivi `tests/test_directions.py`.
5. Aggiorna questo file (`docs/development.md`) e `docs/modules.md`.
6. `pytest` + `ruff check`.
### Principi
- **Foglia pura sopra orchestratore.** Solo `cli.py` può dipendere da
più moduli. Mai il contrario.
- **No mutazione cross‑module.** Ogni modulo possiede il suo stato.
- **Mai dipendere dal payload GMCP da mapper o deadsouls.** Sono
moduli di testo puro.
- **Mai import `cli` da moduli bassi.** Solo `cli` sa di `prompt_toolkit`.
---
## Debugging tipico
| Sintomo | Tool |
| --- | --- |
| IAC parser perso un byte | `pytest tests/test_telnet.py -v` + aggiungere print in `_byte()` |
| GMCP module non riconosciuto | leggere `gmcp.parse_gmcp` con `data=b"..."` in REPL |
| Stanze non rilevate senza GMCP | stampare `state.parser._buf` dopo feed_text |
| Mapper collassa in `[mappa troppo larga]` | abbassare `max_cols`/`max_rows` in `render(...)` |
| Stanza GMCP arriva ma render non aggiorna | `state.mapper.current_room` punta ancora al vecchio? |
| Password finisce nel log | grep `--log-md` per `hunter2`/`***` ratio |
| Sessione hung dopo quit | `conn.close()` chiamato? `finally` cleanup? |
---
## Aggiungere un nuovo comando slash
1. In `cli.py:_handle_local_command`, aggiungi un ramo `elif cmd == "foo":`.
2. Se modifica stato → metti la logica prima del `if state.logger is not None: state.logger.local_cmd(...)`.
3. Se inoltra al MUD → usa `await conn.send_line("foo")`.
4. Aggiungi voce nella tabella di `docs/usage.md` e `_print_help()`.
5. Test: aggiungi in `tests/test_cli.py` (futuro; ancora non esiste).
---
## Versioning
`__version__` in `__init__.py` (`0.1.0` in v0.2). Bump:
- **patch** (0.1.x): bug fix, no nuovi flag.
- **minor** (0.x.0): nuovi flag, nuove feature senza break della CLI.
- **major** (x.0.0): break di `--log-*` schema eventi o rename simboli
pubblici dei moduli.
---
## CI minima (Gitea Actions, opzionale)
Esempio `.gitea/workflows/ci.yml`:
```yaml
name: CI
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.10"
- run: python -m pip install -e ".[dev]"
- run: pytest
- run: ruff check src tests
```
Adatta ai vincoli del tuo runner Gitea (alcuni runner non hanno
`actions/setup-python@v5`; pinning a versioni disponibili è tipico).
+209
View File
@@ -0,0 +1,209 @@
# Schema eventi del `EventLogger`
`EventLogger` espone due sink paralleli attivabili via `__init__`:
- **JSONL** (`--log-json PATH`) — uno `{"ts":..., "kind":..., "data":{...}}` per riga.
- **Markdown** (`--log-md PATH`) — heading `## HH:MM:SS.mmm — kind` seguito da
un body specifico del `kind`.
Entrambi `flush()` dopo ogni evento (sub‑os buffer vuoto) → adatto a
`tail -f`.
---
## Timestamp
- **JSONL**: ISO‑8601 UTC con precisione millisecondo + `Z` finale
(es. `"2026-07-05T12:34:56.789Z"`).
- **MD**: solo l'ora, `HH:MM:SS.mmm`.
Entrambi calcolati da `datetime.now(timezone.utc)` non da `time.time()`
(locale‑indipendente).
---
## Tipi di evento
### `session_start` (1 per sessione)
```json
{"kind": "session_start", "data": {"host": "localhost", "port": 6666, "user": ""}}
```
MD: header `# MUD log — <ts>` con `host`, `port`, `user` poi `---`.
### `session_end` (1 per sessione)
```json
{"kind": "session_end", "data": {"reason": "done"}}
```
Possibili `reason`: `done`, `quit`.
### `disconnect`
```json
{"kind": "disconnect", "data": {"reason": "local_close"}}
```
Emesso quando `conn.close()` viene chiamato explicitamente
(`--quit`, EOF, ecc.).
### `text`
```json
{"kind": "text", "data": {"raw": "Inn Hall\n\x1b[36mCobbles.\x1b[0m\n[ Exits: n ]\n"}}
```
Chunk ANSI grezzo dal server, subito dopo la IAC strip. Nel MD:
blocco ` ``` ` con testo privato di ANSI via `strip_ansi`.
> Attenzione: ANSI con byte non validi vengono preservati nel `raw`
> JSONL ma al massimo appaiono come "?" nel markdown. La redazione
> non si applica al text (se il server ti invia una password in
> chiaro la vedrai).
### `room_enter`
```json
{"kind": "room_enter", "data": {"id": "vnum12345", "name": "Inn Hall", "exits": ["n", "e"]}}
```
`id`:
- `vnum{int(num)}` se GMCP `Room.Info.num`/`vnum`/`id` era numerico.
- altrimenti `name.lower()` (case‑insensitive identity).
MD: lista bullet `id` / `name` / `exits`.
### `gmcp`
```json
{"kind": "gmcp", "data": {"module": "Char.Vitals", "payload": {"hp": 100, "max_hp": 120}}}
```
`payload` è il JSON decodificato (potrebbe essere `null` se il modulo
veniva emesso senza body). MD: blocco ` ```json ` con `payload` ordinato
per chiave + indentazione.
### `vitals`
```json
{"kind": "vitals", "data": {"hp": 100, "max_hp": 120, "sp": 50, "max_sp": 60, "mv": 80, "max_mv": 100}}
```
Solo i campi effettivamente presenti nel payload GMCP. MD:
`HP 100/120 — SP 50/60 — MV 80/100`.
### `line_sent`
```json
{"kind": "line_sent", "data": {"text": "north", "redacted": false}}
{"kind": "line_sent", "data": {"text": "***", "redacted": true}}
```
`redacted=true` ⇒ la `text` diventa `"***"` (la stringa originale
viene scartata). MD: `> … *(redacted)*` se redatto, altrimenti `> …`.
### `local_cmd`
```json
{"kind": "local_cmd", "data": {"cmd": "map", "args": [], "redacted": false}}
{"kind": "local_cmd", "data": {"cmd": "password", "args": ["***"], "redacted": true}}
```
Se `redacted=true`, `args` è collassato a una lista di `"***"` di pari
lunghezza all'originale (anche password multi‑parola).
### `subneg` (forward hook diagnostico)
```json
{"kind": "subneg", "data": {"opt": "NAWS", "bytes": 4}}
{"kind": "subneg", "data": {"opt": "TTYPE", "bytes": 0}}
```
Emesso da `cli.py` per ogni subnegoziazione telnet **non‑GMCP**
ricevuta. `opt` è il nome umano (`ECHO`/`SUPPRESS_GO_AHEAD`/
`TTYPE`/`NAWS`/`GMCP` o `opt_<num>` se sconosciuto) da `_OPT_NAMES`,
`bytes` è la lunghezza del payload. Utile in `tail -f` per
diagnosticare la negoziazione delle opzioni senza dover attivare
`--log DEBUG`.
MD: `- **opt**: NAWS` + `- **bytes**: 4`.
### `login`
```json
{"kind": "login", "data": {"phase": "user_sent", "value": "alice", "redact": false}}
{"kind": "login", "data": {"phase": "password_sent", "value": "***", "redact": true}}
{"kind": "login", "data": {"phase": "password_prompt", "value": "", "redact": true}}
```
`phase ∈ {user_sent, password_prompt, password_sent}`.
Default `redact=True` ⇒ `value` registrato come `"***"`. Passa
`redact=False` esplicitamente per i nomi account (devono essere
tracciabili per leggere i log).
MD: `**{phase}** → \`{value}\`` (con back‑tick escape).
---
## Esempio di sessione (estratti)
```bash
$ jq -c '.' < session.jsonl
{"ts":"2026-07-05T12:34:56.789Z","kind":"session_start","data":{"host":"localhost","port":6666,"user":"zork"}}
{"ts":"2026-07-05T12:34:57.012Z","kind":"text","data":{"raw":"Welcome to Dead Souls.\n"}}
{"ts":"2026-07-05T12:34:57.234Z","kind":"text","data":{"raw":"What name do you wish?\n"}}
{"ts":"2026-07-05T12:34:57.235Z","kind":"line_sent","data":{"text":"zork","redacted":false}}
{"ts":"2026-07-05T12:34:57.236Z","kind":"login","data":{"phase":"user_sent","value":"zork","redact":false}}
{"ts":"2026-07-05T12:34:58.001Z","kind":"text","data":{"raw":"Password:\n"}}
{"ts":"2026-07-05T12:34:58.002Z","kind":"login","data":{"phase":"password_prompt","value":"","redact":true}}
{"ts":"2026-07-05T12:35:00.500Z","kind":"line_sent","data":{"text":"***","redacted":true}}
{"ts":"2026-07-05T12:35:00.501Z","kind":"login","data":{"phase":"password_sent","value":"***","redact":true}}
{"ts":"2026-07-05T12:35:01.123Z","kind":"gmcp","data":{"module":"Room.Info","payload":{"num":12345,"name":"Town Square","exits":{"n":12346}}}}
{"ts":"2026-07-05T12:35:01.124Z","kind":"room_enter","data":{"id":"vnum12345","name":"Town Square","exits":["n"]}}
{"ts":"2026-07-05T12:35:30.000Z","kind":"session_end","data":{"reason":"quit"}}
```
---
## Helpers di analisi (`jq`)
```bash
# solo stanze visitate
jq -c 'select(.kind=="room_enter")' session.jsonl
# vitals (ultimo snapshot)
jq -c 'select(.kind=="vitals")' session.jsonl | tail -1
# righe inviate (escludendo redacted)
jq -c 'select(.kind=="line_sent" and .data.redacted==false)' session.jsonl
# tutti i comandi locali slash
jq -c 'select(.kind=="local_cmd")' session.jsonl
```
---
## Idempotenza
`EventLogger.close()` può essere chiamato più volte senza effetti
collaterali. Nel `finally` di `cli._run`, viene chiamato esattamente
una volta anche se l'utente fa `Ctrl‑C` due volte.
---
## Note di privacy
- Il client **non** logga la password in chiaro. Tutti i punti di
emissione relativi a credenziali sono protetti da `redacted=True`.
- `text` chunk può contenere **qualsiasi cosa** il server ti invii. Se
il MUD fa echo della tua password (raro, perché WILL ECHO è attivo),
apparirà nel log.
- `gmcp.payload` è decodificato; se un modulo MUD contiene segreti, li
vedrai nel log.
- Per ambienti condivisi si consiglia `--log-json` su FS cifrato e/o
log ruotato con uno scheduler esterno (es. `logrotate`).
+199
View File
@@ -0,0 +1,199 @@
# Pubblicazione su Gitea
Server: **http://192.168.0.183:8085**
Owner: `truetype`
Repo: `progetto2`
Branch iniziale: `main`
---
## ⚠️ Sicurezza del token
Il token Gitea in colonna `truetype` (stringa 40‑hex) fornisce accesso
**completo** all'account e ai suoi repo. Regole non negoziabili:
1. **Non committare mai** il token nel repo. Il comando di push
sotto lo passa *inline* nell'URL di una sola `git push`. Mai
`git config credential.helper`, mai `.netrc` committato, mai
variabili d'ambiente in uno script versionato.
2. **Revoca dopo l'uso.** Dopo il primo push riuscito, vai su
`http://192.168.0.183:8085/user/settings/applications` → revoca il
token usato. Generane uno nuovo *solo* se ti serve un altro push.
Conserva il vecchio token solo se prevedi push automatici da CI
(allora mettilo nei **secrets** della pipeline, non nel codice).
3. **Preferisci token *scoped***: quando ne crei uno nuovo, limita
`repository.write` e basta, non `repository.admin`.
4. **HTTPS vs SSH.** Se hai una chiave SSH sull'account Gitea,
preferisci sempre SSH: `git@192.168.0.183:8085:truetype/progetto2.git`
(nota: Gitea usa `:owner/repo.git` dopo il `:`). Nessun token in
chiaro è necessario.
> I log di esempio in questo file usano il placeholder `<TOKEN>`.
> **Non** sostituirlo durante la lettura: il token reale dovrebbe
> vivere solo nella shell history dell'operatore.
---
## Creazione del repo (una tantum)
```bash
# 1: crea il repo via API (body JSON minimo)
curl -X POST http://192.168.0.183:8085/api/v1/user/repos \
-H "Authorization: token <TOKEN>" \
-H "Content-Type: application/json" \
-d '{"name": "progetto2", "description": "MUD client (Dead Souls / LP-MUD)", "auto_init": false, "private": false}'
```
Risposta attesa: `201 Created` con JSON contenente `id`, `ssh_url`,
`clone_url`, `html_url`. Annota:
- `html_url` (visibile in browser)
- `clone_url` (es. `http://192.168.0.183:8085/truetype/progetto2.git`)
---
## Configurazione git locale (una tantum)
Da `/home/ubuntu/code/progetto2`:
```bash
git config user.name "truetype"
git config user.email "truetype@192.168.0.183"
```
(Non mettere `user.email` reale se non vuoi che appaia nei commit.)
---
## Primo push
Se `.git/` non esiste ancora:
```bash
cd /home/ubuntu/code/progetto2
git init
git branch -M main
```
Commit iniziale (esclude venv, pycache, cache pytest via `.gitignore`
già presente):
```bash
git add .
git commit -m "Initial commit: MUD client v0.2 (cli, telnet, gmcp, mapper, deadsouls, logger, tests)"
```
Push (token inline, **non** salvato):
```bash
git push https://truetype:<TOKEN>@192.168.0.183:8085/truetype/progetto2.git main
```
Verifica:
```bash
git remote -v
# deve mostrare origin → http://192.168.0.183:8085/truetype/progetto2.git
git log --oneline -5
```
Dopo il push, **revoca `<TOKEN>`** come da sezione sicurezza.
### Strip delle credenziali dalla config locale
Dopo un push con URL `https://user:token@host/...`, Gitea lascia
**l'URL completo (con token!)** scritto in `.git/config`. Rimuovi
subito le credenziali:
```bash
git remote set-url origin http://192.168.0.183:8085/truetype/progetto2.git
git remote -v
# deve mostrare: origin http://192.168.0.183:8085/truetype/progetto2.git (no user:token@)
```
Da questo momento i push successivi richiederanno il token solo se
usi HTTPS nudo. Per evitare di digitarlo ogni volta, configura una
**chiave SSH** (`http://192.168.0.183:8085/user/settings/keys`) e
cambia il remote in `git@192.168.0.183:8085:truetype/progetto2.git`.
---
## Push successivi (senza token)
Dopo aver configurato `git remote add origin ...` con un URL senza
credenziali (es. SSH o HTTPS nudo), i push successivi sono:
```bash
git add .
git commit -m "Fix: ..."
git push origin main
```
Se devi restare su HTTPS, configura un **credential helper**
*temporaneo* nella shell (niente su disco):
```bash
GIT_ASKPASS=/bin/true git -c credential.helper= -c core.askPass=/bin/true push origin main
```
(in pratica è più comodo usare SSH.)
---
## Branch & tag
```bash
# feature branch
git checkout -b feature/msdp-support
git push -u origin feature/msdp-support
# tag di release
git tag -a v0.2.0 -m "release v0.2.0"
git push origin v0.2.0
```
I tag abilitano la releases page di Gitea; usa i tag `vX.Y.Z` come nomi
dei binari / delle release notes.
---
## Re‑push dopo perdita di `.git/`
Se per qualunque motivo il `.git/` locale viene perso:
```bash
cd /home/ubuntu/code/progetto2
git init
git branch -M main
git remote add origin https://192.168.0.183:8085/truetype/progetto2.git
git add .
git commit -m "Restore: progetto2 v0.2"
git push -u origin main
# NB: questo push richiede di nuovo un token se SSH non è configurato
```
---
## Webhook di release (opzionale)
Per triggerare un deploy o un CI su tag push:
1. In Gitea: `Repo → Settings → Webhooks → Add webhook`
2. URL: il tuo endpoint CI (es. Drone, Woodpecker, Gitea Actions)
3. Events: `Push events` + `Create events` (quest'ultimo copre i tag)
4. Secret: `<WEBHOOK_SECRET>` (salvalo nei secrets della CI, non nel
codebase).
---
## Troubleshooting
| Errore | Causa | Rimedio |
| --- | --- | --- |
| `403 Forbidden` | token errato o revocato | rigenera; controlla che lo scope includa `repo` |
| `401 Unauthorized` | token mancante o scaduto | verifica header `Authorization` |
| `repository does not exist` | repo non creato lato Gitea | ricrea via API; assicurati che `owner`/`name` siano giusti |
| `failed to push some refs` | remote ahead | `git pull --rebase origin main` poi push |
| `Could not resolve host` | Gitea offline | controlla `curl http://192.168.0.183:8085/api/v1/version` |
| `Permission denied (publickey)` | SSH non configurato | aggiungi la chiave pubblica a `http://192.168.0.183:8085/user/settings/keys` |
+376
View File
@@ -0,0 +1,376 @@
# Moduli — tour ragionato
Un modulo per sezione. Le funzioni e le classi elencate sono quelle
dell'API *pubblica* del modulo — leaky implementation details sono
omesse (sono `_underscore_prefixed` e cambiano spesso).
---
## `ansi.py` — stripper ANSI
**Scopo.** Funzione pura per togliere gli escape ANSI/VT100 prima che il
testo venga dato in pasto a euristiche di parsing (stanze, login prompt,
mapper UI). È una foglia del grafo: non importa nulla e non sa nulla
del MUD.
**API pubblica.**
```python
def strip_ansi(text: str) -> str: ...
def strip_ansi_bytes(data: bytes) -> bytes: ...
def visible_len(text: str) -> int: ...
```
**Internals.** Una regex (`_ANSI_RE`) compilata a import‑time che
copre CSI, OSC, charset designation e DEC private mode. Sostituzione
`re.sub` lineare. I single ESC byte senza sequenza sono lasciati
intatti (non sono errori, sono escape non validi che la stampante
gestirà).
---
## `deadsouls.py` — euristiche Dead Souls / LP‑MUD
**Scopo.** Riconoscere stanze dal testo ricevuto dal server e
canonicalizzare le direzioni delle uscite. *Solo* testo e solo
euristica — nessuna conoscenza di IAC, GMCP, asyncio.
**API pubblica.**
```python
EXIT_ALIASES: dict[str, str]
DIRECTION_OFFSETS: dict[str, tuple[int, int]]
OPPOSITE: dict[str, str] # importato da mapper
def parse_exits(line: str) -> list[str]
@dataclass
class RoomBlock:
name: str
body: str
exits: list[str]
def parse_room_block(block: str) -> RoomBlock | None
class RollingRoomParser:
def __init__(self) -> None
def feed_text(self, text: str) -> list[RoomBlock]
def flush(self) -> list[RoomBlock]
```
**Strategia di parsing blocco stanza.** Basata su linee (non regex
multi‑linea):
1. Accumula testo in un buffer interno.
2. Aggiungi il nuovo testo.
3. Split per `\n`; conserva l'ultimo pezzo parziale come buffer.
4. Per ogni riga completa: se contiene `Exits:` → chiudi il blocco
corrente (nome = prima riga non vuota, body = resto) ed emetti un
`RoomBlock`. Altrimenti appendi al blocco corrente.
5. Il blocco corrente rimane bufferizzato per la prossima chiamata.
**Tolleranze.** ANSI già strippato a monte. Linee con la parola
"exits" in mezzo a dialoghi NPC sono al sicuro: `parse_exits`
controlla il *formato* (bracketed/unbracketed, ":" o "-") prima di
popolare la lista uscite.
---
## `gmcp.py` — parser GMCP
**Scopo.** Decodificare il payload GMCP (`IAC SB GMCP <module> [<json>]
IAC SE`) in `GmcpEvent(module, data)`. Validare la *forma* del module
name. Estrarre i campi tolleranti (sinonimi) di `Char.Vitals` /
`Room.Info`.
**API pubblica.**
```python
@dataclass(frozen=True)
class GmcpEvent:
module: str
data: Any
# moduli standard
MODULE_CHAR_LOGIN: str
MODULE_CHAR_LOGOUT: str
MODULE_CHAR_VITALS: str
MODULE_CHAR_STATUS: str
MODULE_CHAR_INFO: str
MODULE_ROOM_INFO: str
MODULE_ROOM_WALK_DIR: str
def parse_gmcp(data: bytes) -> GmcpEvent | None
def extract_vitals(payload: Any) -> dict[str, int]
def extract_room_info(payload: Any) -> tuple[str, str, str, str | None] | None
```
**Sinonimi tollerati (vitale).**
| Chiave canonica | Sinonimi accettati (case‑sensitive) |
| --- | --- |
| `hp` / `max_hp` | `hp`, `health`, `life` |
| `sp` / `max_sp` | `sp`, `mana`, `magic` |
| `mv` / `max_mv` | `mv`, `move`, `stam` |
**Room.Info.** Il primo argomento è l'`id` stabile della stanza: se
`num`/`vnum`/`id` è presente e numerico, uso `vnum{int(num)}`;
altrimenti `name.strip().lower()`. Questo è ciò che permette al mapper
di ri‑identificare la stessa stanza tra sessioni.
---
## `mapper.py` — grafo stanze + render ASCII
**Scopo.** Persistenza in‑memory delle stanze visitate e rendering
ASCII del grafo. Le coordinate vengono assegnate **solo** al momento
del render (lazy layout).
**API pubblica.**
```python
CELL_W: int = 22
@dataclass
class Room:
id: str
name: str
body: str
exits: dict[str, str] # direzione -> id vicino
class Mapper:
def __init__(self) -> None
def visit(self, name, body, exits, *, id_override=None) -> str
def link_rooms(self, from_id, direction, to_id) -> None
def move(self, direction: str) -> bool
@property
def current_room(self) -> Room | None
@property
def rooms(self) -> dict[str, Room]
def render(self, *, max_cols=12, max_rows=16) -> str
```
**Algoritmo di layout (3 fasi).**
1. **BFS dal current.** Anchor `(0, 0)` ed esplora i forward edges
noti. Su collisione, nudge `+1` su `y` (fino a 16 tentativi).
2. **Orphan placement.** Per ogni stanza non ancora piazzata, prova
un edge in *qualsiasi direzione* verso un placed‑room (forward o
reverse). Itera finché non ci sono più avanzamenti. Questo gestisce
il backtracking `n → s` dove la `s`‑room non era collegata via
`exits[e]` ma tramite il movimento utente.
3. **Stack in basso a destra.** Le stanze veramente isolate finiscono
qui. Se ne trovi mai una, hai probabilmente un bug di parsing: ogni
stanza raggiungibile ha almeno un edge.
**Snapshot in `_place_adjacent`.** `placed = list(coords.items())` viene
copiato *prima* di mutare `coords` (altrimenti l'iterazione successiva
vedrebbe il nuovo elemento appena inserito e potrebbe creare cicli).
**Render.** Coordinate y ordinate top‑down (y grande = nord). Celle di
larghezza fissa `CELL_W = 22`. Marker `»` per la stanza corrente, `·`
per le altre. I link `e → w` sono tratteggi `───`, le verticali `│`.
---
## `logger.py` — `EventLogger`
**Scopo.** Sink append‑only per eventi di sessione. Due formati
paralleli: JSON Lines (machine‑readable) e Markdown (human‑readable).
File aperti in write‑mode (non append): ogni sessione riparte pulito.
**API pubblica.**
```python
class EventLogger:
def __init__(
self,
json_path: str | None = None,
md_path: str | None = None,
*,
host: str = "",
port: int = 0,
user: str = "",
) -> None
def close(self) -> None
def event(self, kind: str, data: dict[str, Any] | None = None) -> None
# shortcut tipizzati
def session_start(self, host: str, port: int, user: str) -> None
def session_end(self, reason: str = "") -> None
def disconnect(self, reason: str = "") -> None
def text(self, raw_ansi: str) -> None
def room(self, rid: str, name: str, exits: list[str]) -> None
def gmcp(self, module: str, payload: Any) -> None
def vitals(
self, *,
hp=None, sp=None, mv=None,
max_hp=None, max_sp=None, max_mv=None,
) -> None
def line_sent(self, line: str, *, redacted: bool = False) -> None
def local_cmd(self, cmd, args=None, *, redacted: bool = False) -> None
def login(self, phase: str, value: str = "", *, redact: bool = True) -> None
```
**Thread safety.** `threading.Lock` interno. Oggi è uncontended
(asyncio è single‑threaded) ma difende da futuri worker thread. Tutte
le scritture passano per `_write_json_unlocked` + `_append_md_unlocked`
+ `_flush_unlocked` sotto il lock.
**Sicurezza password.** Tre punti di redazione:
- `line_sent(line, redacted=True)` → testo registrato come `"***"`.
- `local_cmd(cmd, args, redacted=True)` → args list collassato a `["***"]`.
- `login(phase, value, redact=True)` (default) → value registrato come `"***"`.
**Idempotenza.** `close()` è idempotente; `__init__` solleva `OSError`
in caso di errore di I/O e chiude gli handle aperti prima di rilanciare
(nessun file descriptor leak).
---
## `telnet.py` — trasporto telnet
**Scopo.** Implementare la state machine IAC (RFC 854) sopra un
`asyncio.StreamReader`/`StreamWriter` e gestire la negoziazione
opzioni (ECHO/NAWS/TTYPE/GMCP). **Nessuna** conoscenza del MUD o della
UI.
**API pubblica.**
```python
# costanti di protocollo
IAC, DO, DONT, WILL, WONT, SB, SE = 255, 253, 254, 251, 252, 250, 240
EOR, GA = 239, 249
OPT_ECHO, OPT_TTYPE, OPT_NAWS, OPT_GMCP = 1, 24, 31, 201
# eventi
@dataclass
class TextEvent: text: str
@dataclass
class NegotiateEvent: cmd: int; opt: int
@dataclass
class SubnegEvent: opt: int; data: bytes
@dataclass
class PromptEvent: ...
TelnetEvent = TextEvent | NegotiateEvent | SubnegEvent | PromptEvent
class IACParser:
"""Push parser: feed bytes, get events."""
def feed(self, buf: bytes) -> list[TelnetEvent]
class TelnetConnection:
def __init__(self, host, port, *,
naws_size=None, ttype="XTERM-256COLOR", want_gmcp=True) -> None
async def connect(self) -> None
async def close(self) -> None
async def read_loop(self, handler: Callable[[TelnetEvent], Awaitable[None]]) -> None
async def send(self, data: bytes) -> None
async def send_line(self, line: str) -> None
async def report_window_size(self, cols: int, rows: int) -> None
```
**State machine (`S` enum).**
```
NORMAL → byte normale → NORMAL
NORMAL → IAC → IAC
IAC → IAC → NORMAL (literal)
IAC → DO|DONT|WILL|WONT → NEGOTIATE
IAC → SB → SB_OPTION
IAC → SE → NORMAL (stray)
IAC → EOR|GA → NORMAL (PromptEvent emesso)
IAC → altro (NOP/DM/AO) → NORMAL
NEGOTIATE → <opt> → NORMAL (NegotiateEvent emesso)
SB_OPTION → <opt> → SB_DATA
SB_DATA → IAC → SB_IAC
SB_IAC → SE → NORMAL (SubnegEvent emesso)
SB_IAC → IAC → SB_DATA (literal)
SB_IAC → altro → NORMAL (abort)
```
**Negoziazione opzioni.**
| cmd | opt | Risposta client |
| --- | --- | --- |
| `DO` | `ECHO`/`NAWS`/`TTYPE`/`GMCP` se supportato | `WILL` (+ side‑effect: invio dimensione finestra per NAWS, set `expects_ttype_send` per TTYPE) |
| `DO` | altro | `WONT` |
| `DONT` | qualsiasi | `WONT` |
| `WILL` | qualsiasi | `DONT` (server offers non ci servono in v0.2) |
| `WONT` | qualsiasi | `DONT` |
**Side‑channel GMCP.** Se `want_gmcp=True`, `connect()` invia
proattivamente `IAC WILL GMCP` prima di leggere. Alcuni MUD preferiscono
l'offerta client piuttosto che `DO` server.
---
## `cli.py` — orchestratore
**Scopo.** Compostare tutti i moduli sopra in un'unica sessione
interattiva. Possiede il prompt, la status bar, il dispatcher telnet,
la macchina a stati di auto‑login e i comandi slash locali.
**API pubblica.**
```python
@dataclass
class ClientState:
mapper: Mapper
parser: RollingRoomParser
vitals: dict[str, int]
gmcp_enabled: bool
auto_user_sent: bool
auto_password_prompted: bool
auto_password_sent: bool
logger: EventLogger | None
def main(argv: list[str] | None = None) -> int
```
**Componenti.**
- `_format_prompt(state, user)` → `FormattedText` con `[room] <user> > `.
- `_format_status_bar(state)` → `FormattedText` con `room`, GMCP
on/off, hp/sp/mv.
- `_handle_gmcp(data, state)` → dispatch modulo → vitale/room/mapper.
- `_maybe_autologin(text, conn, state, args)` → matching regex su
`What name do you wish?` / `Password:` e invio automatico.
- `_on_telnet_event(ev, conn, state, args)` → dispatcher principale.
- `_input_loop(session, conn, state, user)` → prompt loop async.
- `_handle_local_command(line, conn, state) -> bool` → comandi slash.
- `_run(args)` → orchestrata con `asyncio.wait(FIRST_COMPLETED)`,
`conn.close()` *prima* del drain delle pending task.
**Comandi slash.**
| Slash | Implementazione | Note |
| --- | --- | --- |
| `/quit` `/exit` `/q` | `conn.close()` + `SystemExit(0)` | non loggato |
| `/help` `/?` | `_print_help()` | loggato |
| `/map` | `mapper.render()` stampato | loggato |
| `/where` | room corrente + uscite | loggato |
| `/look` | `conn.send_line("look")` | loggato |
| `/stats` | stanze, GMCP, vitali | loggato |
| `/vitals` | `state.vitals` | loggato |
| `/clear` | `prompt_toolkit.clear()` | loggato |
| `/password PWD...` | redacted + `conn.send_line(pwd)` | loggato (redatto) |
| altro `/...` | `return False` → forward al MUD | loggato come `line_sent` da `_input_loop` |
**Auto‑login state machine.**
```
text: "What name do you wish?"
↓
args.user exists & not auto_user_sent
↓
send_line(args.user); auto_user_sent = True
↓
text: "Password:"
↓
auto_password_prompted = True
├── args.password exists → send_line, auto_password_sent = True
└── args.password "" → stampa suggerimento /password PWD
```
+177
View File
@@ -0,0 +1,177 @@
# Protocolli gestiti
Il client parla **IAOC telnet** (RFC 854) con un subset scelto di
opzioni e **GMCP** come estensione JSON. Niente MSDP/MCCP/MXP in v0.2
(`WONT` su richiesta server).
---
## Telnet — state machine
I byte IAC (`0xFF`) sono interpretati come command o iniziatore di
subneg. Tutto il resto è testo che viene accumulato e poi emesso come
un unico `TextEvent` per chunk letto.
```
byte new state effetto
─────────────────────────────────────────────────────────────────────
0xFF IAC nessuno
0xFF 0xFF NORMAL literal 0xFF nel testo
0xFF DO|DONT|WILL|WONT → NEGOTIATE parser aspetta option byte
0xFF SB → SB_OPTION parser aspetta option byte
0xFF SE NORMAL drop silenzioso
0xFF EOR NORMAL + emit PromptEvent
0xFF GA NORMAL + emit PromptEvent
0xFF + altro NORMAL drop silent (NOP/DM/AO/AYT…)
```
In subneg:
```
byte new state effetto
─────────────────────────────────────────────────────────────────────
SB_OPTION: <opt> SB_DATA
SB_DATA: non IAC accumula in _sb_buf
SB_DATA: IAC SB_IAC
SB_IAC : SE NORMAL + emit SubnegEvent(opt, _sb_buf)
SB_IAC : IAC SB_DATA + literal 0xFF in _sb_buf
SB_IAC : altro NORMAL abort della subneg
```
`EOR` (End Of Record, 239) e `GA` (Go Ahead, 249) sono entrambi
interpretati come **prompt boundary**. Dead Souls li usa per
delimitare l'output tra una stanza e l'altra. Il client li propaga
come `PromptEvent`. Nella v0.2 nessun consumer in `cli.py` lo
ascolta (`elif isinstance(ev, PromptEvent): return None` è un no‑op
esplicito): è un **forward hook** per future euristiche di stanza
basate sul confine EOR/GA.
### Negoziazione opzioni
| Server chiede | Client risponde | Side effect |
| --- | --- | --- |
| `DO ECHO` | `WILL ECHO` | il prompt locale provvede l'eco |
| `DO NAWS` (con `naws_size`) | `WILL NAWS` + `SB NAWS w h SE` | dimensione iniziale |
| `DO TTYPE` (con `ttype`) | `WILL TTYPE` | set `_expects_ttype_send` |
| `DO TTYPE` (SB SEND) | `SB TTYPE "XTERM-256COLOR" SE` | risposta |
| `DO GMCP` (con `want_gmcp`) | `WILL GMCP` | abilita dispatch eventi GMCP |
| `DO` altro | `WONT` | rifiuto gentile |
| `DONT X` | `WONT X` | sempre |
| `WILL X` | `DONT X` | server offers non accettati |
| `WONT X` | `DONT X` | sempre |
### GMCP — WILL proattivo
`TelnetConnection(want_gmcp=True)` invia `IAC WILL GMCP` al
`connect()`. Questo è richiesto da alcuni MUD (Dead Souls incluso) che
aspettano l'offerta del client invece di un `DO GMCP` server‑driven.
---
## GMCP — wire format
GMCP è un subneg con option code `201`. Il payload è testo UTF‑8:
```
IAC SB GMCP <module.path> [' ' <json>] IAC SE
```
Esempio:
```
IAC SB GMCP Char.Vitals {"hp":100,"max_hp":120,"sp":42,"mv":80} IAC SE
```
### Parsing
In `gmcp.parse_gmcp`:
1. `data.decode('utf-8', errors='replace')` (tollerante a byte non validi).
2. `lstrip()` per edge case con junk iniziale.
3. Split sul primo spazio: `head, _, rest = text.partition(' ')`.
4. Validazione `head` con `_MODULE_NAME_RE = ^[A-Za-z][A-Za-z0-9_.]*$`:
rifiuta `.bad`, `1Room`, `Room..Info`, ecc.
5. `json.loads(rest)` se presente. Decode error → `data = None`
(tolleranza: server potrebbe dribblare payload parziali).
### Moduli supportati out‑of‑the‑box
| Modulo | Effetto |
| --- | --- |
| `Char.Vitals` | aggiorna `state.vitals` con sinonimi (`hp`/`health`/`life`) |
| `Room.Info` | `mapper.visit()` con `id_override = "vnum{int(num)}"` se `num`/`vnum`/`id` è numerico |
| `Char.Login`, `Char.Logout`, `Char.Status`, `Char.Info` | loggati (futuri hook) |
| `Room.WalkDir` | loggato (futuro hook per movimento automatico) |
Vedi `docs/events.md` per i payload registrati.
---
## Dead Souls — euristiche di stanza
Dead Souls (driver FluffOS) usa convenzione di stanza testuale:
```
Inn Hall
A warm fire crackles in the hearth.
[ Exits: n e ]
```
Il parser `RollingRoomParser` riconosce:
1. **Nome**: prima riga non vuota del blocco corrente.
2. **Body**: tutte le righe tra nome e uscite (incluso l'eventuale
riga `[ Exits: ... ]` nel body del `RoomBlock`).
3. **Exits**: l'ultima riga del blocco riconosciuta dal regex
`_EXITS_RE` (bracketed/unbracketed, `:` o `-`).
4. **`No exits`**: la regex `_NO_EXITS_RE` rende la stanza "isolata"
senza abortire.
`parse_exits(line)` è tollerante:
```
Exits: north, south
[ Exits: n e ]
Exit -
[ Exit: none ]
```
Tutti questi formati producono la stessa `list[str]` canonicalizzata.
---
## ANSI — stripper
`ansi.strip_ansi(text)` rimuove sequenze CSI, OSC, charset designation
e DEC private mode. Sequenze non riconosciute vengono lasciate nel
testo (meglio un carattere spurio che un evento perso).
Pattern coperti:
```regex
\x1B\[[0-9;?]*[ -/]*[@-~] # CSI sequences (SGR, cursor, mode)
| \x1B\][^\x07\x1B]*(?:\x07|\x1B\\) # OSC sequences (title, etc.)
| \x1B[()][A-Za-z0-9] # Charset designation
| \x1B\[?[0-9;]*[hl] # DEC private mode set/reset
```
**Single ESC byte** senza `[` successivo è lasciato nel testo. I log
JSON e MD sono protetti da un `try/except` difensivo nel logger che,
in caso di crash dello stripper, registra il raw (meglio spurio che
event perso).
---
## Error handling sui byte non validi
| Input | Comportamento |
| --- | --- |
| 0xFF in mezzo al testo (literal) | preservato nel testo |
| Subneg interrotta | `SubnegEvent` mai emesso, stato torna a NORMAL; ok |
| `IAC` con comando sconosciuto | drop silent |
| UTF‑8 non valido | `errors='replace'` su tutto il testo |
| GMCP JSON malformato | `GmcpEvent.data = None` (non abort) |
| GMCP module name malformato | `parse_gmcp` ritorna `None` (non propagato) |
| GMCP module name non‑riconosciuto | loggato come evento "gmcp" ma non dispatch |
| Stdout errno dopo read | `IncompleteReadError` / `ConnectionResetError` → chiusura pulita |
+194
View File
@@ -0,0 +1,194 @@
# Uso della CLI — riferimento completo
Il comando dopo `pip install -e .` è **`mudclient`** (entry point
definito in `pyproject.toml`). È equivalente a `python -m progetto2`.
---
## Flag
| Flag | Tipo | Default | Note |
| --- | --- | --- | --- |
| `--host HOST` | str | `localhost` | host MUD (default punta al docker locale di Dead Souls) |
| `--port PORT` | int | `6666` | porta TCP |
| `--user NAME` | str | `""` (vuoto = no auto‑login del nome) | auto‑login quando il server chiede "What name do you wish?" |
| `--password PWD` | str | `os.environ.get("MUDCLIENT_PASSWORD", "")` | password per auto‑login dopo `Password:` |
| `--cols COLS` | int | `120` | larghezza iniziale finestra (NAWS subneg) |
| `--rows ROWS` | int | `40` | altezza iniziale finestra |
| `--log LVL` | choice | `WARNING` | livello Python logging. Valori validi (rifiutati se diversi): `DEBUG` / `INFO` / `WARNING` / `ERROR` / `CRITICAL`. |
| `--log-json PATH` | str | `""` (no log JSON) | path JSONL append‑only (vedi `docs/events.md`) |
| `--log-md PATH` | str | `""` (no log MD) | path Markdown streamed |
> **Sicurezza `--password`**: il valore è in chiaro in `ps`/`history`.
> Usare l'env `MUDCLIENT_PASSWORD` o `/password PWD` (redatto). Il
> client redige automaticamente il valore nei log JSON/MD.
---
## Esempi rapidi
```bash
# Connessione minimale al docker di sviluppo
mudclient
# Server remoto esplicito
mudclient --host mud.example.com --port 4000
# Auto‑login completo (account + password)
mudclient --user zork --password hunter2
# password via env (preferibile in CI / shell history):
MUDCLIENT_PASSWORD=hunter2 mudclient --user zork
# Auto‑login solo nome; password manuale quando il server chiede
mudclient --user zork
# Finestra ridotta (es. 80x24) per terminali piccoli
mudclient --cols 80 --rows 24
# Debug del parser / IAC
mudclient --log DEBUG --log-json ./debug.jsonl --log-md ./debug.md
# Log persistente di una sessione intera (rotation esterna)
mudclient --log-json ~/mud-$(date +%Y%m%d-%H%M).jsonl \
--log-md ~/mud-$(date +%Y%m%d-%H%M).md
```
---
## Comandi slash (locali)
| Slash | Implementazione | Loggato come |
| --- | --- | --- |
| `/help` o `/?` | stampa aiuto | `local_cmd` |
| `/map` | render ASCII del mapper | `local_cmd` |
| `/where` | "Sei in: {Room}. Uscite: {list}" | `local_cmd` |
| `/look` | `conn.send_line("look")` | `local_cmd` |
| `/stats` | locali scoperti, GMCP, vitali | `local_cmd` |
| `/vitals` | `state.vitals` raw | `local_cmd` |
| `/clear` | `prompt_toolkit.clear()` | `local_cmd` |
| `/quit` `/exit` `/q` | chiude + SystemExit(0) | non loggato |
| `/password PWD` | redacted + `conn.send_line(pwd)` | `local_cmd` (redatto) + `login` |
| `/...` (altro) | passa al MUD con `/` rimosso | `line_sent` |
`Ctrl-D` chiude in modo equivalente a `/quit`.
---
## Prompt e status bar
**Prompt** (ricostruito a ogni line):
```
[NomeStanza]> _
```
`[NomeStanza]` cyan brillante, `>` verde brillante.
**Status bar** (ricostruito a ogni refresh):
```
[stanza: Outer Courtyard] [GMCP: on] HP 100/100 SP 50/50 MV 80/80
```
Refresh forzato ogni **0.5 s** via `PromptSession(refresh_interval=0.5)`.
Costo: ~2 redraw/sec anche in idle. Documentato come scelta di
trade‑off tra reattività e CPU nel file `docs/architecture.md`.
---
## Flusso auto‑login
```
server: "What name do you wish? " (o "Enter your name:", "Login:")
client: send_line(args.user) ← solo se --user e one-shot
client: state.auto_user_sent = True
server: "Password: "
client: state.auto_password_prompted = True
├── args.password → send_line(args.password); state.auto_password_sent = True
└── args.password "" → stampa: "[login] server chiede password; digita /password <pwd>"
```
`auto_*_sent` è **one‑shot**: una seconda occorrenza di "What name…?"
non re‑invio il nome. Previene spam in caso di messaggi di echo o
re‑auth lato server.
**Varianti regex coperte** (`_LOGIN_RE`):
```
What name do you wish
Enter your name
Login: / Login name
Name to use
```
più prefissi opzionali `>` o `*`.
**Varianti password** (`_PASSWORD_RE`):
```
Password:
* Password ?
> Password:
```
Tolleranti a prefissi `> *`, spaziature e `: ?` finale.
Se il tuo MUD usa prompt diversi, modifica le regex in
`src/progetto2/cli.py` (sono 3 righe ciascuna).
---
## Output delle stanze
Dead Souls con GMCP attivo invia `Room.Info {num, name, desc, exits}`.
Il client usa `num` come id stabile della stanza. Senza GMCP, le
stanze sono rilevate euristicamente vedendo:
```
Inn Hall
A warm fire crackles.
[ Exits: n e ]
```
`RollingRoomParser` chiude il blocco su `Exits:` line. Il mapper
visita la stanza immediatamente. Puoi digitare `n` o `e` e la prossima
volta che il server conferma la stanza, **l'edge viene creato**.
---
## Rotazione dei log
I path `--log-json` e `--log-md` vengono **truncati** all'apertura.
Per conservare le sessioni precedenti:
```bash
mv ~/mud.jsonl{,.prev} && mudclient --log-json ~/mud.jsonl
```
Per archivi automatici prima di ogni sessione:
```bash
LOGBASE=~/mud-$(date +%Y%m%d-%H%M)
mudclient --log-json "$LOGBASE.jsonl" --log-md "$LOGBASE.md"
```
Per tail in tempo reale:
```bash
# in un secondo terminale durante una sessione attiva
tail -f ~/mud.jsonl | jq -c '.kind + ": " + (.data|tostring)'
```
---
## Diagnosi minimali
| Sintomo | Probabile causa | Cosa provare |
| --- | --- | --- |
| Il client si blocca dopo Connect | Server MUD non risponde `DO ECHO` / GMCP | `--log DEBUG` + `--log-json ./x.jsonl` |
| Prompt non ridisegnato | `refresh_interval` non rinfresca in idle | è normale, aspettare il prossimo input |
| `/map` mostra [mappa vuota] | nessuna stanza ancora riconosciuta | cammina un po', oppure controlla se `Room.Info` GMCP arriva |
| Login password non parte | regex `_PASSWORD_RE` non match | `--log DEBUG` e ispezione `data` di `text` events |
| Sessione termina subito | `conn.close()` per EOF inatteso | `--log INFO` per vedere l'eccezione in `read_loop` |
+49
View File
@@ -0,0 +1,49 @@
[project]
name = "progetto2"
version = "0.1.0"
description = "CLI MUD client + utilities (Dead Souls / LP-MUD over telnet)"
readme = "README.md"
requires-python = ">=3.10"
license = { text = "MIT" }
authors = [{ name = "Author" }]
keywords = ["mud", "telnet", "dead-souls", "lp-mud", "cli"]
classifiers = [
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3 :: Only",
"Operating System :: OS Independent",
"Environment :: Console",
"Intended Audience :: End Users/Desktop",
"Topic :: Games/Entertainment",
"Topic :: Internet :: Telnet",
]
dependencies = [
"prompt_toolkit>=3.0.43",
]
[project.optional-dependencies]
dev = [
"pytest>=7",
"ruff>=0.5",
]
[project.scripts]
mudclient = "progetto2.cli:main"
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
include = ["progetto2*"]
[tool.setuptools.package-dir]
"" = "src"
[tool.pytest.ini_options]
addopts = "-q"
testpaths = ["tests"]
[tool.ruff]
line-length = 100
target-version = "py310"
+3
View File
@@ -0,0 +1,3 @@
"""progetto2 — CLI MUD client + utilities."""
__version__ = "0.1.0"
+7
View File
@@ -0,0 +1,7 @@
"""Allow ``python -m progetto2`` to launch the CLI."""
from __future__ import annotations
from progetto2.cli import main
if __name__ == "__main__":
raise SystemExit(main())
+32
View File
@@ -0,0 +1,32 @@
"""Minimal ANSI / VT100 escape-sequence utilities.
The display layer uses :func:`prompt_toolkit.formatted_text.ANSI` to render
coloured text directly. The parser layer (room detection, mapper) operates on
plain text, so we need a fast ANSI stripper.
"""
from __future__ import annotations
import re
# Comprehensive-enough stripper: CSI sequences (incl. SGR / cursor moves),
# OSC sequences, charset designation, and DEC private mode set/reset.
_ANSI_RE = re.compile(
r"\x1B\[[0-9;?]*[ -/]*[@-~]"
r"|\x1B\][^\x07\x1B]*(?:\x07|\x1B\\)"
r"|\x1B[()][A-Za-z0-9]"
r"|\x1B\[?[0-9;]*[hl]"
)
def strip_ansi(text: str) -> str:
"""Return ``text`` with ANSI/VT100 escape sequences removed."""
return _ANSI_RE.sub("", text)
def strip_ansi_bytes(data: bytes) -> bytes:
return _ANSI_RE.sub(b"", data)
def visible_len(text: str) -> int:
"""Length of ``text`` once ANSI escapes are removed (used for layout)."""
return len(strip_ansi(text))
+544
View File
@@ -0,0 +1,544 @@
"""Entry point for the ``mudclient`` script.
Usage:
mudclient [--host HOST] [--port PORT] \
[--user NAME] [--password PWD] \
[--cols COLS] [--rows ROWS]
Defaults to ``localhost:6666`` (Dead Souls docker container). Outputs the
remote text rendered with ANSI codes, offers a small set of slash-prefixed
local commands (``/map``, ``/help``, ``/quit``, ``/vitals``, …) before
sending raw text to the MUD. Negotiates NAWS / TTYPE / ECHO / GMCP via the
:class:`TelnetConnection`.
"""
from __future__ import annotations
import argparse
import asyncio
import logging
import os
import re
import shlex
import sys
from dataclasses import dataclass, field
from prompt_toolkit import PromptSession, print_formatted_text
from prompt_toolkit.formatted_text import ANSI, FormattedText
from prompt_toolkit.patch_stdout import patch_stdout
from prompt_toolkit.shortcuts import clear
from prompt_toolkit.styles import Style
from .ansi import strip_ansi
from .deadsouls import RollingRoomParser
from .gmcp import (
MODULE_CHAR_VITALS,
MODULE_ROOM_INFO,
extract_room_info,
extract_vitals,
parse_gmcp,
)
from .logger import EventLogger
from .mapper import DIRECTION_SHORTCUTS, Mapper
from .telnet import (
OPT_GMCP,
PromptEvent,
SubnegEvent,
TelnetConnection,
TelnetEvent,
TextEvent,
)
logger = logging.getLogger(__name__)
# Friendly names for telnet subnegotiation options (RFC 854 + common
# MUD extensions). Used purely for diagnostic logging so a human
# reader can grep `--log-json` by option name.
_OPT_NAMES: dict[int, str] = {
1: "ECHO",
3: "SUPPRESS_GO_AHEAD",
24: "TTYPE",
31: "NAWS",
201: "GMCP",
}
PROMPT_STYLE = Style.from_dict(
{
"room": "ansibrightcyan bold",
"status": "ansiyellow",
"pulse": "ansibrightgreen",
"vital_hp": "ansibrightred",
"vital_sp": "ansibrightblue",
"vital_mv": "ansigreen",
"toolbar": "reverse",
"info": "ansiwhite",
"warn": "ansibrightyellow",
"error": "ansibrightred bold",
}
)
@dataclass
class ClientState:
"""Mutable per-session state shared between telnet reader and input loop."""
mapper: Mapper
parser: RollingRoomParser
vitals: dict[str, int] = field(default_factory=dict)
gmcp_enabled: bool = False
auto_user_sent: bool = False
auto_password_prompted: bool = False
auto_password_sent: bool = False
logger: EventLogger | None = None
@property
def room_name(self) -> str:
room = self.mapper.current_room
return room.name if room else "(sconosciuto)"
# ---- prompt + status bar formatters --------------------------------------
def _format_prompt(state: ClientState, user: str) -> FormattedText:
parts: list[tuple[str, str]] = [
("class:room", f"[{state.room_name}]"),
("", " "),
]
if user:
parts.append(("class:status", f"<{user}>"))
parts.append(("", " "))
parts.append(("class:pulse", "> "))
return FormattedText(parts)
def _format_status_bar(state: ClientState) -> FormattedText:
parts: list[tuple[str, str]] = [
("class:info", f"[stanza: {state.room_name}]"),
("", " "),
("class:info", f"[GMCP: {'on' if state.gmcp_enabled else 'off'}]"),
]
v = state.vitals
if v:
parts.append(("", " "))
if "hp" in v:
parts.append(
("class:vital_hp", f"HP {v['hp']}/{v.get('max_hp', '?')}")
)
if "sp" in v:
parts.append(("", " "))
parts.append(
("class:vital_sp", f"SP {v['sp']}/{v.get('max_sp', '?')}")
)
if "mv" in v:
parts.append(("", " "))
parts.append(
("class:vital_mv", f"MV {v['mv']}/{v.get('max_mv', '?')}")
)
return FormattedText(parts)
def _status_bar_getter(state: ClientState):
def _get() -> FormattedText:
return _format_status_bar(state)
return _get
# ---- GMCP dispatch -------------------------------------------------------
async def _handle_gmcp(data: bytes, state: ClientState) -> None:
ev = parse_gmcp(data)
if ev is None:
return
if state.logger is not None:
state.logger.gmcp(ev.module, ev.data)
module = ev.module
if module == MODULE_CHAR_VITALS:
v = extract_vitals(ev.data)
if v:
state.vitals.update(v)
if state.logger is not None:
state.logger.vitals(
hp=v.get("hp"),
sp=v.get("sp"),
mv=v.get("mv"),
max_hp=v.get("max_hp"),
max_sp=v.get("max_sp"),
max_mv=v.get("max_mv"),
)
elif module == MODULE_ROOM_INFO:
info = extract_room_info(ev.data)
if info is None:
return
rid, name, body, num_id = info
# Try to read exits via Room.Info if present.
exits_list: list[str] = []
if isinstance(ev.data, dict) and "exits" in ev.data:
raw_exits = ev.data["exits"]
from .deadsouls import EXIT_ALIASES
if isinstance(raw_exits, dict):
for d in raw_exits:
d_canon = EXIT_ALIASES.get(str(d).lower(), str(d).lower())
if d_canon not in exits_list:
exits_list.append(d_canon)
elif isinstance(raw_exits, (list, tuple)):
from .deadsouls import EXIT_ALIASES
for d in raw_exits:
d_canon = EXIT_ALIASES.get(str(d).lower(), str(d).lower())
if d_canon not in exits_list:
exits_list.append(d_canon)
state.mapper.visit(
name=name,
body=body,
exits=exits_list,
id_override=num_id,
)
if state.logger is not None:
state.logger.room(rid, name, exits_list)
# ---- auto-login detection (best-effort) ---------------------------------
_LOGIN_RE = re.compile(
r"(?im)^\s*[>*]?\s*"
r"(?:what\s+name\s+do\s+you\s+wish"
r"|enter\s+(?:your\s+)?name"
r"|login(?:\s+name)?"
r"|name\s+to\s+use"
r")\s*[:?]?\s*$"
)
_PASSWORD_RE = re.compile(r"(?im)^\s*[>*]?\s*password\s*[:?]?\s*$")
async def _maybe_autologin(
text_plain: str,
conn: TelnetConnection,
state: ClientState,
args: argparse.Namespace,
) -> None:
if args.user and not state.auto_user_sent and _LOGIN_RE.search(text_plain):
await conn.send_line(args.user)
state.auto_user_sent = True
if state.logger is not None:
# Username is logged plaintext (both events) — it's the
# identifier the session is bound to and the user already
# typed it on the command line.
state.logger.line_sent(args.user, redacted=False)
state.logger.login("user_sent", value=args.user, redact=False)
print_formatted_text(ANSI(f"\x1b[36m[login] inviato nome: {args.user}\x1b[0m\n"))
if (
state.auto_user_sent
and not state.auto_password_sent
and not state.auto_password_prompted
and _PASSWORD_RE.search(text_plain)
):
state.auto_password_prompted = True
if args.password:
await conn.send_line(args.password)
state.auto_password_sent = True
if state.logger is not None:
# Password is redacted in both line_sent and login events
# so the value never lands in plaintext on disk.
state.logger.line_sent(args.password, redacted=True)
state.logger.login("password_sent", value=args.password)
print_formatted_text(ANSI("\x1b[36m[login] password inviata\x1b[0m\n"))
else:
if state.logger is not None:
state.logger.login("password_prompt")
print_formatted_text(
ANSI(
"\x1b[33m[login] server chiede password; "
"digita /password <pwd>\x1b[0m\n"
)
)
# ---- telnet event dispatcher --------------------------------------------
async def _on_telnet_event(
ev: TelnetEvent,
conn: TelnetConnection,
state: ClientState,
args: argparse.Namespace,
) -> None:
if isinstance(ev, TextEvent):
if ev.text:
if state.logger is not None:
state.logger.text(ev.text)
print_formatted_text(ANSI(ev.text), end="")
chunk_plain = strip_ansi(ev.text or "")
if chunk_plain:
rooms = state.parser.feed_text(chunk_plain)
for r in rooms:
prev = state.mapper.current_room
state.mapper.visit(name=r.name, body=r.body, exits=r.exits)
# Only emit `room_enter` when the room changed (avoid
# noise on re-renders of the same room).
rid_now = state.mapper.current_room.id if state.mapper.current_room else None
if state.logger is not None and (
prev is None or prev.id != rid_now
):
state.logger.room(rid_now or "", r.name, r.exits)
await _maybe_autologin(chunk_plain, conn, state, args)
elif isinstance(ev, SubnegEvent):
if ev.opt == OPT_GMCP:
state.gmcp_enabled = True
await _handle_gmcp(ev.data, state)
elif state.logger is not None:
# Surface other subnegotiations (NAWS, TTYPE, ECHO) with a
# human-readable name so logs are greppable by option.
opt_name = _OPT_NAMES.get(ev.opt, f"opt_{ev.opt}")
state.logger.event(
"subneg",
{"opt": opt_name, "bytes": len(ev.data or b"")},
)
elif isinstance(ev, PromptEvent):
return None
# ---- local commands ------------------------------------------------------
async def _input_loop(
session: PromptSession,
conn: TelnetConnection,
state: ClientState,
user: str,
) -> None:
while True:
try:
line = await session.prompt_async(_format_prompt(state, user))
except (EOFError, KeyboardInterrupt):
await conn.close()
return
if line is None:
continue
line = line.rstrip()
if not line:
continue
if line.startswith("/"):
handled = await _handle_local_command(line, conn, state)
if handled is False:
if state.logger is not None:
state.logger.line_sent(line[1:])
await conn.send_line(line[1:])
continue
# Proactively inform the mapper about likely movements.
if line.lower().split()[0] in DIRECTION_SHORTCUTS:
state.mapper.move(line)
if state.logger is not None:
state.logger.line_sent(line)
await conn.send_line(line)
async def _handle_local_command(
line: str,
conn: TelnetConnection,
state: ClientState,
) -> bool:
"""Return True if handled, False if the line should be sent to the MUD.
On handled branches a single ``local_cmd`` event is emitted (with
``redacted=True`` for ``/password``); fall-through commands don't
get a ``local_cmd`` event because ``_input_loop`` already records
them as ``line_sent`` once they reach the MUD.
"""
try:
tokens = shlex.split(line[1:])
except ValueError as e:
print_formatted_text(ANSI(f"\x1b[31m{line[1:]!r} non valido: {e}\x1b[0m\n"))
return True
if not tokens:
return True
cmd, *args = tokens
if cmd in ("quit", "exit", "q"):
await conn.close()
raise SystemExit(0)
if cmd in ("help", "?"):
_print_help()
elif cmd == "map":
print_formatted_text(ANSI("\x1b[36m" + state.mapper.render() + "\x1b[0m\n"))
elif cmd == "look":
await conn.send_line("look")
elif cmd == "clear":
clear()
elif cmd == "where":
room = state.mapper.current_room
if room is None:
print_formatted_text(ANSI("\x1b[33m(posizione sconosciuta)\x1b[0m\n"))
else:
exits = ", ".join(sorted(room.exits))
print_formatted_text(
ANSI(
f"\x1b[36mSei in: {room.name}\x1b[0m "
f"\x1b[33mUscite: {exits or 'nessuna'}\x1b[0m\n"
)
)
elif cmd == "stats":
print_formatted_text(
ANSI(
f"\x1b[36mLocali scoperti: {len(state.mapper.rooms)} "
f"GMCP: {'on' if state.gmcp_enabled else 'off'} "
f"Vitali: {state.vitals or 'non disponibili'}\x1b[0m\n"
)
)
elif cmd == "vitals":
print_formatted_text(ANSI(f"\x1b[36mVitali: {state.vitals or 'non ricevuti'}\x1b[0m\n"))
elif cmd == "password":
if not args:
# No password supplied — fall through to the wire so the
# server can decide what to do (e.g. ``/password`` itself
# is unknown to it and may bounce back as text).
return False
pwd = " ".join(args)
if state.logger is not None:
# ``/password`` logs everything explicitly with ``***``
# redaction; the generic emit at the bottom of this
# function would otherwise leak the password plaintext.
state.logger.local_cmd("password", args, redacted=True)
state.logger.line_sent(pwd, redacted=True)
state.logger.login("password_sent", value=pwd)
await conn.send_line(pwd)
state.auto_password_prompted = True
state.auto_password_sent = True
print_formatted_text(ANSI("\x1b[36m[login] password inviata\x1b[0m\n"))
return True
else:
# Unrecognised /command → fall through to the MUD after stripping
# the slash. ``_input_loop`` will record it as ``line_sent``.
return False
# Generic log: a non-password local command handled above.
if state.logger is not None:
state.logger.local_cmd(cmd, args)
return True
def _print_help() -> None:
out = ANSI(
"\x1b[36mComandi locali:\x1b[0m\n"
" /help mostra questo messaggio\n"
" /map disegna la mappa ASCII delle stanze visitate\n"
" /where mostra la stanza corrente e le uscite\n"
" /look invia 'look' al MUD\n"
" /vitals mostra hp/sp/mv ricevuti via GMCP\n"
" /stats statistiche locali (stanze + GMCP)\n"
" /password PWD invia una password al MUD\n"
" /clear pulisce lo schermo\n"
" /quit esce dal client\n"
"\nTutto ciò che non inizia con '/' è inviato al MUD.\n"
)
print_formatted_text(out)
# ---- entry point ---------------------------------------------------------
async def _on_telnet_event_factory(state, conn, args):
async def handler(ev: TelnetEvent) -> None:
await _on_telnet_event(ev, conn, state, args)
await conn.read_loop(handler)
async def _run(args: argparse.Namespace) -> int:
state = ClientState(mapper=Mapper(), parser=RollingRoomParser())
state.logger = EventLogger(
json_path=args.log_json or None,
md_path=args.log_md or None,
host=args.host,
port=args.port,
user=args.user,
)
conn = TelnetConnection(
args.host,
args.port,
naws_size=(args.cols, args.rows),
ttype="XTERM-256COLOR",
want_gmcp=True,
)
await conn.connect()
state.logger.session_start(args.host, args.port, args.user)
print_formatted_text(
ANSI(f"\x1b[36mConnesso a {args.host}:{args.port} (GMCP attivo)\x1b[0m\n")
)
session = PromptSession(
style=PROMPT_STYLE,
bottom_toolbar=_status_bar_getter(state),
refresh_interval=0.5, # force redraw so vitals updates are visible
)
reader_task = asyncio.create_task(_on_telnet_event_factory(state, conn, args))
inputer_task = asyncio.create_task(_input_loop(session, conn, state, args.user))
exit_code = 0
exit_reason = "done"
try:
try:
await asyncio.wait(
{reader_task, inputer_task},
return_when=asyncio.FIRST_COMPLETED,
)
except SystemExit as e:
exit_code = int(getattr(e, "code", 0) or 0)
exit_reason = "quit"
# Closing the connection unblocks `reader.read()` if it is mid-await,
# so the reader task can exit via EOF instead of relying on cancellation.
await conn.close()
if state.logger is not None:
state.logger.disconnect(reason="local_close")
for t in (reader_task, inputer_task):
if not t.done():
t.cancel()
try:
await t
except (SystemExit, asyncio.CancelledError, Exception):
pass
return exit_code
finally:
if state.logger is not None:
state.logger.session_end(reason=exit_reason)
state.logger.close()
def _build_argparser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="mudclient",
description="CLI MUD client for Dead Souls / LP-MUD.",
)
p.add_argument("--host", default="localhost")
p.add_argument("--port", type=int, default=6666)
p.add_argument("--user", default="", help="nome personaggio per auto-login")
p.add_argument(
"--password",
default=os.environ.get("MUDCLIENT_PASSWORD", ""),
help="password (o via env MUDCLIENT_PASSWORD)",
)
p.add_argument("--cols", type=int, default=120)
p.add_argument("--rows", type=int, default=40)
p.add_argument(
"--log",
default="WARNING",
choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
)
p.add_argument(
"--log-json",
default="",
help="path to a JSON-Lines session log (one event per line, append-only)",
)
p.add_argument(
"--log-md",
default="",
help="path to a Markdown transcript log (human-readable, append-only)",
)
return p
def main(argv: list[str] | None = None) -> int:
args = _build_argparser().parse_args(argv)
logging.basicConfig(level=getattr(logging, args.log))
try:
return asyncio.run(_run(args))
except KeyboardInterrupt:
return 130
if __name__ == "__main__":
sys.exit(main())
+194
View File
@@ -0,0 +1,194 @@
"""Dead Souls (LP-MUD, FluffOS driver) text heuristics.
Used by the CLI layer to detect room transitions, extract room names, and
canonicalise exit directions. Pure functions — no I/O.
The parser is deliberately conservative: when it can't tell, it returns
``None`` and the caller keeps accumulating.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
# ---- direction aliases and offsets --------------------------------------
EXIT_ALIASES: dict[str, str] = {
"n": "n", "north": "n",
"s": "s", "south": "s",
"e": "e", "east": "e",
"w": "w", "west": "w",
"u": "u", "up": "u",
"d": "d", "down": "d",
"ne": "ne", "northeast": "ne",
"nw": "nw", "northwest": "nw",
"se": "se", "southeast": "se",
"sw": "sw", "southwest": "sw",
"out": "out", "enter": "enter",
}
DIRECTION_OFFSETS: dict[str, tuple[int, int]] = {
"n": (0, -1),
"s": (0, 1),
"e": (1, 0),
"w": (-1, 0),
"u": (0, 0), # vertical not represented on 2-D map
"d": (0, 0),
"ne": (1, -1),
"nw": (-1, -1),
"se": (1, 1),
"sw": (-1, 1),
}
_EXITS_RE = re.compile(
r"(?:^|\n)" # start of line
r"\s*[\[<(\s]?" # optional leading bracket
r"\s*" # whitespace between bracket and keyword
r"[Ee]xits?:?\s*[:\-=]?\s*" # 'Exit:', 'Exits:', 'Exits -'
r"([^]\n\r\)\>]*)" # capture exits payload
r"[\]>)']?",
re.MULTILINE,
)
_NO_EXITS_RE = re.compile(r"\b(?:no\s+exits|none)\b", re.IGNORECASE)
# A line that contains an 'Exits' prompt (used to delimit room blocks).
_EXITS_LINE_RE = re.compile(r"\b[Ee]xits?\b")
def parse_exits(line: str) -> list[str]:
"""Return canonical exit directions found in ``line``.
The function is forgiving: bracketed or unbracketed, "Exits:" or "Exit:",
comma- or space-separated, full names or abbreviations.
"""
m = _EXITS_RE.search(line)
if not m:
return []
raw = m.group(1)
parts = re.split(r"[\s,/]+", raw.strip())
out: list[str] = []
for p in parts:
p_norm = p.strip().lower().rstrip(".,;:!?\"'")
if not p_norm:
continue
if p_norm in EXIT_ALIASES:
d = EXIT_ALIASES[p_norm]
if d not in out:
out.append(d)
continue
# attempt fuzzy: "north-east", "northeast.", etc.
cleaned = p_norm.replace("-", "").replace("_", "")
if cleaned in EXIT_ALIASES:
d = EXIT_ALIASES[cleaned]
if d not in out:
out.append(d)
return out
# ---- room block model + parser ------------------------------------------
@dataclass
class RoomBlock:
name: str
body: str
exits: list[str]
def parse_room_block(block: str) -> RoomBlock | None:
"""Heuristically split a text block into :class:`RoomBlock`.
Conventionally the first non-empty line is the room name and a line
containing 'Exits:' carries the exits list. Description body is the rest.
"""
name = ""
exits: list[str] = []
body_lines: list[str] = []
for ln in block.splitlines():
if "exits" in ln.lower():
exits = parse_exits(ln)
body_lines.append(ln)
continue
if not name and ln.strip():
name = ln.strip()
else:
body_lines.append(ln)
if not name:
return None
if not exits and _NO_EXITS_RE.search(block):
exits = []
elif not exits:
return None
return RoomBlock(name=name, body="\n".join(body_lines).strip(), exits=exits)
class RollingRoomParser:
"""Line-based room-block detector.
Strategy:
* Append incoming text to a buffer.
* Split the buffer into lines.
* Maintain a "current block" list of lines since the last emitted room.
* As soon as we encounter a line containing the word 'Exits:' (matched
via the exits regex), we close the current block: first non-empty line
is the room name, the rest is the body, and we emit a :class:`RoomBlock`.
* Anything left in the current block stays buffered for the next call.
"""
def __init__(self) -> None:
self._buf: str = ""
def feed_text(self, text: str) -> list[RoomBlock]:
rooms: list[RoomBlock] = []
self._buf += text
# Split buffer into lines; the trailing partial line stays in self._buf.
lines = self._buf.split("\n")
self._buf = lines.pop() if lines else ""
cur: list[str] = []
for ln in lines:
if _EXITS_RE.search(ln):
# The 'exits' line closes the current block.
block_lines = cur + [ln]
room = _build_room_from_lines(block_lines)
if room is not None:
rooms.append(room)
cur = []
else:
cur.append(ln)
# Anything left in ``cur`` is the start of the next block; keep it.
self._buf = ("\n".join(cur) + ("\n" + self._buf if self._buf else ""))
return rooms
def flush(self) -> list[RoomBlock]:
rooms = self.feed_text("")
# Best-effort parse of any remaining partial block.
if self._buf.strip():
block = parse_room_block(self._buf)
if block is not None:
rooms.append(block)
self._buf = ""
return rooms
def _build_room_from_lines(block_lines: list[str]) -> RoomBlock | None:
"""Build a RoomBlock from a list of lines that ends with an 'Exits:' line."""
name = ""
body_lines: list[str] = []
exits: list[str] = []
for ln in block_lines:
if "exits" in ln.lower():
exits = parse_exits(ln)
body_lines.append(ln)
continue
if not name and ln.strip():
name = ln.strip()
elif name:
body_lines.append(ln)
if not name:
return None
if not exits and _NO_EXITS_RE.search("\n".join(block_lines)):
exits = []
elif not exits:
return None
return RoomBlock(name=name, body="\n".join(body_lines).strip(), exits=exits)
+146
View File
@@ -0,0 +1,146 @@
"""GMCP (Generic MUD Communication Protocol) parser.
GMCP is a JSON-over-telnet extension:
IAC SB GMCP <module.path> [<json>] IAC SE
where ``<module.path>`` is a dotted identifier (e.g. ``Char.Vitals``) and
the optional payload after a single space is a JSON document (object,
array, or scalar).
This module only parses — it does not negotiate (that's the transport's
job). It is called from the CLI layer whenever the :class:`SubnegEvent`
option number is :data:`telnet.OPT_GMCP`.
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from typing import Any
from .deadsouls import EXIT_ALIASES
# Module-name shape: leading alpha + word chars / dots / underscores.
_MODULE_NAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.]*$")
# Widely-used GMCP module identifiers.
MODULE_CHAR_LOGIN = "Char.Login"
MODULE_CHAR_LOGOUT = "Char.Logout"
MODULE_CHAR_VITALS = "Char.Vitals"
MODULE_CHAR_STATUS = "Char.Status"
MODULE_CHAR_INFO = "Char.Info"
MODULE_ROOM_INFO = "Room.Info"
MODULE_ROOM_WALK_DIR = "Room.WalkDir"
@dataclass(frozen=True)
class GmcpEvent:
module: str # dotted module id, e.g. 'Char.Vitals'
data: Any # parsed JSON value or None if payload absent/invalid
def parse_gmcp(data: bytes) -> GmcpEvent | None:
"""Parse an IAC SB payload from a GMCP subnegotiation.
Returns ``None`` for empty / non-GMCP frames. ``data=None`` is
returned when only the module name is present (no JSON body).
Module names must match ``[A-Za-z][A-Za-z0-9_.]*`` (a leading
letter followed by alphanumerics / dots / underscores). This
rejects malformed frames such as ``".bad"`` or ``"1Room"``.
"""
try:
text = data.decode("utf-8", errors="replace")
except (UnicodeDecodeError, AttributeError):
return None
text = text.lstrip()
if not text:
return None
head, _, rest = text.partition(" ")
module = head.strip()
if not module or not _MODULE_NAME_RE.match(module):
return None
parsed: Any = None
rest = rest.strip()
if rest:
try:
parsed = json.loads(rest)
except json.JSONDecodeError:
# Some servers dribble partial payloads — tolerate silently.
parsed = None
return GmcpEvent(module=module, data=parsed)
def extract_vitals(payload: Any) -> dict[str, int]:
"""Tolerantly extract hp / sp / mv from a ``Char.Vitals`` payload.
Recognised synonyms (case-sensitive in the payload):
hp ← hp / health / life
sp ← sp / mana / magic
mv ← mv / move / stam
Returns whatever was found. Each entry is only added if the *current*
value is a number or numeric string.
"""
if not isinstance(payload, dict):
return {}
out: dict[str, int] = {}
def as_int(*keys: str) -> int | None:
for k in keys:
v = payload.get(k)
if isinstance(v, bool):
continue
if isinstance(v, (int, float)):
return int(v)
if isinstance(v, str) and v.isdigit():
return int(v)
return None
for short, syns in (
("hp", ("hp", "health", "life")),
("sp", ("sp", "mana", "magic")),
("mv", ("mv", "move", "stam")),
):
cur = as_int(*syns)
if cur is not None:
out[short] = cur
mx = as_int("max_" + syns[0], "max_" + syns[1], "max_" + syns[2])
if mx is not None:
out["max_" + short] = mx
return out
def extract_room_info(payload: Any) -> tuple[str, str, str, str | None] | None:
"""Extract (id, name, body, num_id_or_None) from a ``Room.Info`` payload.
``num_id_or_None`` is the server-assigned vnum (or None).
The returned id is vnum-based when a vnum is present, else a
stable hash of the (lowercased) name for casing-insensitive identity.
"""
if not isinstance(payload, dict):
return None
name = payload.get("name") or payload.get("title")
if not isinstance(name, str) or not name.strip():
return None
num_id = payload.get("num") or payload.get("id") or payload.get("vnum")
name_clean = name.strip()
if isinstance(num_id, (int, float)):
rid = f"vnum{int(num_id)}"
else:
rid = name_clean.lower()
body = payload.get("desc") or payload.get("description") or ""
exits_raw = payload.get("exits", {})
exits: list[str] = []
if isinstance(exits_raw, dict):
for d in exits_raw:
d_canon = EXIT_ALIASES.get(str(d).lower(), str(d).lower())
if d_canon not in exits:
exits.append(d_canon)
elif isinstance(exits_raw, (list, tuple)):
for d in exits_raw:
d_canon = EXIT_ALIASES.get(str(d).lower(), str(d).lower())
if d_canon not in exits:
exits.append(d_canon)
return rid, name_clean, str(body), rid if num_id is not None else None
+331
View File
@@ -0,0 +1,331 @@
"""Append-only session logger for the MUD client.
Captures every event the client receives from the server, dispatches
to the user input loop, or infers from in-memory state. Events are
serialised as JSON Lines (one event per line, UTF-8) and/or streamed
into a human-readable Markdown transcript. Both sinks ``flush()`` after
each write so ``tail -f`` and live editors see updates while the
session is running.
Invocation is opt-in: pass paths via ``--log-json PATH`` and/or
``--log-md PATH``.
JSONL schema (one line per event)::
{"ts": "2026-07-05T12:34:56.789Z",
"kind": "session_start",
"data": {"host": "localhost", "port": 6666, "user": "foo"}}
Event kinds
-----------
``session_start`` connection established (host/port/user)
``session_end`` normal termination
``disconnect`` socket closed (with optional reason)
``text`` one raw ANSI chunk from the server
``room_enter`` mapper recorded a visit (fires on first arrive AND revisit)
``gmcp`` a parsed IAC SB GMCP frame (module + decoded JSON payload)
``vitals`` hp / sp / mv snapshot update
``line_sent`` user keystroke forwarded to the MUD (passwords redacted)
``local_cmd`` intercepted ``/...`` command
``login`` auto-login prompt / response
"""
from __future__ import annotations
import datetime as _dt
import json
import os
import threading
from typing import Any, IO
from .ansi import strip_ansi
def _utcnow_iso() -> str:
"""ISO8601 UTC with millisecond precision and trailing ``Z``."""
return (
_dt.datetime.now(_dt.timezone.utc)
.isoformat(timespec="milliseconds")
.replace("+00:00", "Z")
)
class EventLogger:
"""Two-sink append-only session logger.
Both files are opened in write mode at construction so a fresh
session starts with a clean file. Each event issues a small write
plus a ``flush()`` so concurrent readers (``tail -f``, ``less +F``,
any editor with auto-reload) see the update promptly without
waiting for the kernel buffer.
Construct with either or both paths. ``None`` disables that sink.
All write operations are synchronised internally so they're safe
to call from multiple coroutines (asyncio is single-threaded, but
defensive locking avoids tearing partial lines if the writer
becomes multi-threaded later).
"""
def __init__(
self,
json_path: str | None = None,
md_path: str | None = None,
*,
host: str = "",
port: int = 0,
user: str = "",
) -> None:
self._json_path = self._expand(json_path)
self._md_path = self._expand(md_path)
self._json_f: IO[str] | None = None
self._md_f: IO[str] | None = None
self._md_buf: list[str] = []
self._closed = False
# asyncio is single-threaded and every Logger method is purely
# synchronous today, so this lock is uncontended. We keep it as
# a defensive measure: if a future caller dispatches logging
# from a worker thread (e.g. a queued writer), the lock stops
# two threads from interleaving partial writes.
self._lock = threading.Lock()
try:
if self._json_path:
self._json_f = open(self._json_path, "w", encoding="utf-8")
if self._md_path:
self._md_f = open(self._md_path, "w", encoding="utf-8")
self._md_write_header(host, port, user)
except OSError:
# If either file open or the header write fails (only IO
# errors are expected here), close any handles we managed
# to open so we don't leak them. Rethrow so the caller can
# surface the failure.
self._close_files()
raise
@staticmethod
def _expand(p: str | None) -> str | None:
if not p:
return None
return os.path.expanduser(p)
# -- public API ------------------------------------------------------
def close(self) -> None:
"""Flush any buffered markdown lines and close both files."""
if self._closed:
return
with self._lock:
self._flush_unlocked()
self._close_files()
self._closed = True
def event(self, kind: str, data: dict[str, Any] | None = None) -> None:
"""Emit a generic event. ``kind`` is the snake_case category."""
if self._closed:
return
obj = {"ts": _utcnow_iso(), "kind": kind, "data": data or {}}
with self._lock:
if self._closed: # re-check under lock
return
self._write_json_unlocked(obj)
self._append_md_unlocked(obj)
self._flush_unlocked()
# -- typed convenience methods --------------------------------------
def session_start(self, host: str, port: int, user: str) -> None:
self.event("session_start", {"host": host, "port": port, "user": user})
def session_end(self, reason: str = "") -> None:
self.event("session_end", {"reason": reason})
def disconnect(self, reason: str = "") -> None:
self.event("disconnect", {"reason": reason})
def text(self, raw_ansi: str) -> None:
self.event("text", {"raw": raw_ansi})
def room(self, rid: str, name: str, exits: list[str]) -> None:
self.event("room_enter", {"id": rid, "name": name, "exits": exits})
def gmcp(self, module: str, payload: Any) -> None:
self.event("gmcp", {"module": module, "payload": payload})
def vitals(
self,
*,
hp: int | None = None,
sp: int | None = None,
mv: int | None = None,
max_hp: int | None = None,
max_sp: int | None = None,
max_mv: int | None = None,
) -> None:
data: dict[str, int] = {}
for short, mx, cur in (
("hp", max_hp, hp),
("sp", max_sp, sp),
("mv", max_mv, mv),
):
if cur is not None:
data[short] = cur
if mx is not None:
data["max_" + short] = mx
self.event("vitals", data)
def line_sent(self, line: str, *, redacted: bool = False) -> None:
self.event(
"line_sent",
{"text": "***" if redacted else line, "redacted": redacted},
)
def local_cmd(
self,
cmd: str,
args: list[str] | None = None,
*,
redacted: bool = False,
) -> None:
# When ``redacted`` is True the entire args list is collapsed
# into one ``***`` marker — protects against multi-word
# passwords like ``/password secret phrase`` showing only the
# first word redacted.
args_list = args or []
rendered_args = ["***" for _ in args_list] if redacted else list(args_list)
self.event(
"local_cmd",
{"cmd": cmd, "args": rendered_args, "redacted": redacted},
)
def login(
self,
phase: str,
value: str = "",
*,
redact: bool = True,
) -> None:
"""Record a login-related event.
``value`` is rendered as ``"***"`` by default to keep passwords
and other secrets out of the log. Pass ``redact=False`` (e.g.
for an account name) when you want the value to be plaintext.
``phase`` and ``value`` accept positional args; ``redact`` is
keyword-only so callers must opt in to plaintext explicitly.
"""
rendered = "***" if (value and redact) else value
self.event(
"login",
{"phase": phase, "value": rendered, "redact": redact},
)
# -- internals -------------------------------------------------------
def _write_json_unlocked(self, obj: dict[str, Any]) -> None:
if self._json_f is None:
return
line = json.dumps(obj, ensure_ascii=False, separators=(", ", ": "))
self._json_f.write(line)
self._json_f.write("\n")
def _append_md_unlocked(self, obj: dict[str, Any]) -> None:
if self._md_f is None:
return
# Time-only stamp ("12:34:56.789") makes a tighter heading.
ts_short = obj["ts"].split("T", 1)[1].rstrip("Z")
kind = obj["kind"]
data = obj.get("data") or {}
self._md_buf.append(f"## {ts_short} — {kind}\n")
if kind == "text":
raw = data.get("raw", "") or ""
# ``strip_ansi`` already swallows lone ESC bytes, but a
# belt-and-braces try keeps a malformed chunk (rare,
# defer to Python's default error handling) from poisoning
# the entire log if a future change makes it stricter.
try:
plain = strip_ansi(raw)
except Exception:
plain = raw
self._md_buf.append("```\n")
for ln in (plain.splitlines() or [""]):
self._md_buf.append(ln)
self._md_buf.append("```\n")
elif kind == "room_enter":
rid = str(data.get("id", ""))
name = str(data.get("name", ""))
exits = list(data.get("exits") or [])
self._md_buf.append(f"- **id**: `{rid}`\n")
self._md_buf.append(f"- **name**: {name}\n")
self._md_buf.append(
f"- **exits**: {', '.join(exits) if exits else '—'}\n"
)
elif kind == "gmcp":
module = str(data.get("module", ""))
payload = data.get("payload")
self._md_buf.append(f"- **module**: `{module}`\n\n")
self._md_buf.append("```json\n")
self._md_buf.append(
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)
)
self._md_buf.append("\n```\n")
elif kind == "vitals":
parts: list[str] = []
for short, mx in (("hp", "max_hp"), ("sp", "max_sp"), ("mv", "max_mv")):
if short in data:
parts.append(
f"{short.upper()} {data[short]}/{data.get(mx, '?')}"
)
self._md_buf.append(" — ".join(parts) or "- (no values)\n")
elif kind == "line_sent":
tag = " *(redacted)*" if data.get("redacted") else ""
escaped = data.get("text", "").replace("`", "\\`")
self._md_buf.append(f"> `{escaped}`{tag}\n")
elif kind == "local_cmd":
cmd = str(data.get("cmd", ""))
args = list(data.get("args") or [])
tag = " *(redacted)*" if data.get("redacted") else ""
args_str = "".join(f" {a}" for a in args)
self._md_buf.append(f"`/{cmd}`{args_str}{tag}\n")
elif kind == "session_start":
self._md_buf.append(f"- **host**: `{data.get('host', '')}:`\n")
self._md_buf.append(f"- **port**: {data.get('port', 0)}\n")
self._md_buf.append(f"- **user**: `{data.get('user', '')}`\n")
elif kind == "session_end":
self._md_buf.append(f"reason: `{data.get('reason', '')}`\n")
elif kind == "disconnect":
self._md_buf.append(f"reason: `{data.get('reason', '')}`\n")
elif kind == "login":
phase = str(data.get("phase", ""))
value = str(data.get("value", ""))
extra = f" → `{value}`" if value else ""
self._md_buf.append(f"**{phase}**{extra}\n")
self._md_buf.append("") # blank-line separator
def _md_write_header(self, host: str, port: int, user: str) -> None:
if self._md_f is None:
return
ts = _utcnow_iso()
self._md_f.write(f"# MUD log — {ts}\n\n")
self._md_f.write(
f"- **host**: `{host}`\n- **port**: {port}\n- **user**: `{user}`\n\n"
)
self._md_f.write("---\n\n")
self._md_f.flush()
def _flush_unlocked(self) -> None:
if self._md_f is not None and self._md_buf:
self._md_f.write("\n".join(self._md_buf))
self._md_f.write("\n")
self._md_buf.clear()
self._md_f.flush()
if self._json_f is not None:
self._json_f.flush()
def _close_files(self) -> None:
for f in (self._json_f, self._md_f):
if f is not None:
try:
f.close()
except Exception:
pass
self._json_f = None
self._md_f = None
+314
View File
@@ -0,0 +1,314 @@
"""In-memory room graph + ASCII map renderer for the MUD client.
The mapper stores rooms as a directed graph keyed by a *stable* ID derived
from the room name and canonical exits (the description is intentionally
ignored because Dead Souls room text mutates with time/weather/items).
When a GMCP ``Room.Info`` package provides a vnum, that is used as the ID
override for stable cross-session identity.
Coordinates are computed lazily at render-time:
1. Anchor ``current`` at ``(0, 0)`` and BFS outward via forward edges.
2. Iteratively place *orphan* rooms adjacent to any placed room they have
a forward or reverse 1-step edge with (handles backtracking: walking
``n`` then ``s``).
3. Truly isolated rooms are stacked in the lower-right corner so the
player never loses them off-screen.
"""
from __future__ import annotations
import hashlib
from collections import deque
from dataclasses import dataclass, field
from .deadsouls import DIRECTION_OFFSETS, EXIT_ALIASES
CELL_W = 22 # room name column width
# Common movement shortcuts the client can detect in user input.
DIRECTION_SHORTCUTS: frozenset[str] = frozenset(
{"n", "s", "e", "w", "ne", "nw", "se", "sw", "u", "d"}
)
OPPOSITE: dict[str, str] = {
"n": "s", "s": "n", "e": "w", "w": "e",
"ne": "sw", "sw": "ne", "nw": "se", "se": "nw",
"u": "d", "d": "u",
}
@dataclass
class Room:
id: str
name: str
body: str
exits: dict[str, str] = field(default_factory=dict) # direction -> neighbour id
class Mapper:
"""Pure-graph mapper. Coordinates are assigned only at render-time."""
def __init__(self) -> None:
self._rooms: dict[str, Room] = {}
self._order: list[str] = []
self._current: str | None = None
self._pending_dir: str | None = None # direction user just attempted
# -- ingestion --------------------------------------------------------
def visit(
self,
name: str,
body: str,
exits: list[str],
*,
id_override: str | None = None,
) -> str:
"""Record a room. Returns its stable id. ``id_override`` lets the
caller (GMCP) force the id to a server-assigned vnum.
"""
normalized = [EXIT_ALIASES.get(x, x) for x in exits]
rid = id_override if id_override else self._stable_id(name, normalized)
if rid in self._rooms:
existing = self._rooms[rid]
existing.body = body
for d in normalized:
existing.exits.setdefault(d, "")
else:
self._rooms[rid] = Room(
id=rid,
name=name.strip(),
body=body,
exits={d: "" for d in normalized},
)
self._order.append(rid)
# Resolve pending movement: link old_current -> new (this id).
if (
self._pending_dir is not None
and self._current is not None
and self._current != rid
):
self._rooms[self._current].exits[self._pending_dir] = rid
self._pending_dir = None
self._current = rid
return rid
def link_rooms(self, from_id: str, direction: str, to_id: str) -> None:
if from_id in self._rooms:
self._rooms[from_id].exits[direction] = to_id
# -- movement ---------------------------------------------------------
def move(self, direction: str) -> bool:
if not self._current or self._current not in self._rooms:
return False
d = EXIT_ALIASES.get(direction.lower(), direction.lower())
if d not in self._rooms[self._current].exits:
self._pending_dir = d
return False
target = self._rooms[self._current].exits[d]
if target and target in self._rooms:
self._current = target
self._pending_dir = None
return True
self._pending_dir = d
return False
@property
def current_room(self) -> Room | None:
if self._current is None:
return None
return self._rooms.get(self._current)
@property
def rooms(self) -> dict[str, Room]:
return self._rooms
# -- stable id --------------------------------------------------------
@staticmethod
def _stable_id(name: str, exits: list[str]) -> str:
canonical = " ".join(sorted({EXIT_ALIASES.get(x, x) for x in exits}))
h = hashlib.sha1(f"{name.strip().lower()}|{canonical}".encode()).hexdigest()[:12]
return h
# -- rendering --------------------------------------------------------
def render(self, *, max_cols: int = 12, max_rows: int = 16) -> str:
if not self._rooms:
return "[mappa vuota]"
if self._current is None or self._current not in self._rooms:
return "[posizione sconosciuta]"
coords = self._assign_coords()
if not coords:
return "[mappa vuota]"
x_span = max(c[0] for c in coords.values()) - min(c[0] for c in coords.values()) + 1
y_span = max(c[1] for c in coords.values()) - min(c[1] for c in coords.values()) + 1
if x_span > max_cols:
return f"[mappa temporaneamente troppo larga ({x_span} > {max_cols} colonne)]"
if y_span > max_rows:
return f"[mappa temporaneamente troppo alta ({y_span} > {max_rows} righe)]"
rows: dict[int, list[tuple[int, str]]] = {}
for rid, (cx, cy) in coords.items():
rows.setdefault(cy, []).append((cx, rid))
for r in rows:
rows[r].sort()
min_y, max_y = min(rows), max(rows)
out_lines: list[str] = []
for y in range(max_y, min_y - 1, -1):
cells = rows.get(y, [])
cell_at_x: dict[int, str] = {x: rid for x, rid in cells}
xs = list(cell_at_x.keys()) or [0]
min_x, max_x = min(xs), max(xs)
buffer = self._render_row(cell_at_x, min_x, max_x)
out_lines.append(buffer.rstrip())
if y > min_y:
out_lines.append(self._render_vertical_row(cell_at_x, min_x, max_x))
return "\n".join(out_lines)
# -- assignment -------------------------------------------------------
def _assign_coords(self) -> dict[str, tuple[int, int]]:
"""Multi-source BFS from ``current`` + iterative orphan placement
+ fallback for truly isolated rooms.
"""
assert self._current is not None
coords: dict[str, tuple[int, int]] = {self._current: (0, 0)}
placed: set[str] = {self._current}
# Phase 1: BFS via forward known edges.
queue: deque[str] = deque([self._current])
while queue:
rid = queue.popleft()
x, y = coords[rid]
for d, target in self._rooms[rid].exits.items():
if not target or target not in self._rooms or target in coords:
continue
nx, ny = self._next_xy(x, y, d, coords)
coords[target] = (nx, ny)
placed.add(target)
queue.append(target)
# Phase 2: orphan placement — any room with a 1-step edge (any
# direction) to a placed room gets placed adjacent to it. Repeat
# until no more can be placed (handles chains).
changed = True
while changed:
changed = False
for rid in list(self._rooms):
if rid in placed:
continue
if self._place_adjacent(rid, coords):
placed.add(rid)
changed = True
# Phase 3: truly isolated rooms — stack in lower-right corner.
max_x_seen = max((c[0] for c in coords.values()), default=0)
max_y_seen = max((c[1] for c in coords.values()), default=0)
base_x = max_x_seen + 2
base_y = max_y_seen
for rid in list(self._rooms):
if rid in placed:
continue
xx, yy = base_x, base_y
while any(c == (xx, yy) for c in coords.values()):
xx += 1
coords[rid] = (xx, yy)
placed.add(rid)
base_x = xx + 1
return coords
def _place_adjacent(self, rid: str, coords: dict[str, tuple[int, int]]) -> bool:
"""Try a forward or reverse edge to a placed room. Returns True on success."""
# Snapshot the placed positions so we don't mutate coords during iteration.
placed = list(coords.items())
# Forward: placed_rid exits[dir] == rid → rid is in direction `dir` from placed_rid.
for placed_rid, (px, py) in placed:
for d, target in self._rooms[placed_rid].exits.items():
if target == rid:
nx, ny = self._next_xy(px, py, d, coords)
coords[rid] = (nx, ny)
return True
# Reverse: rid exits[dir] == placed_rid. So rid is in OPPOSITE[dir] from placed_rid.
for placed_rid, (px, py) in placed:
for d, target in self._rooms[rid].exits.items():
if target == placed_rid:
opposite = OPPOSITE.get(d)
if not opposite:
continue
nx, ny = self._next_xy(px, py, opposite, coords)
coords[rid] = (nx, ny)
return True
return False
def _next_xy(
self,
x: int,
y: int,
direction: str,
coords: dict[str, tuple[int, int]],
) -> tuple[int, int]:
"""Compute (x + dx, y + dy) for ``direction``, nudging south on collision."""
dx, dy = DIRECTION_OFFSETS.get(direction, (0, 0))
nx, ny = x + dx, y + dy
shifted = 0
while any(c == (nx, ny) for c in coords.values()) and shifted < 16:
ny += 1
shifted += 1
return nx, ny
# -- helpers ----------------------------------------------------------
def _render_row(
self, cell_at_x: dict[int, str], min_x: int, max_x: int
) -> str:
parts: list[str] = []
for x in range(min_x, max_x + 1):
rid = cell_at_x.get(x)
if rid is None:
parts.append(" " * CELL_W)
else:
room = self._rooms[rid]
marker = "»" if rid == self._current else "·"
text = f"{marker} {self._truncate(room.name, CELL_W - 3)}"
parts.append(text.ljust(CELL_W))
if x < max_x:
left = cell_at_x.get(x)
right = cell_at_x.get(x + 1)
connected = (
left is not None
and right is not None
and self._is_east_linked(left, right)
)
parts.append("─── " if connected else " ")
return "".join(parts)
def _render_vertical_row(
self, cell_at_x: dict[int, str], min_x: int, max_x: int
) -> str:
parts: list[str] = []
for x in range(min_x, max_x + 1):
char = "│" if x in cell_at_x else " "
parts.append(char.center(CELL_W + 4))
return "".join(parts).rstrip()
def _is_east_linked(self, left_id: str, right_id: str) -> bool:
left = self._rooms.get(left_id)
right = self._rooms.get(right_id)
if left is None or right is None:
return False
return left.exits.get("e") == right_id or right.exits.get("w") == left_id
@staticmethod
def _truncate(text: str, width: int) -> str:
if width <= 0:
return ""
if len(text) <= width:
return text
return text[: width - 1] + "…"
+292
View File
@@ -0,0 +1,292 @@
"""Async telnet transport with IAC state machine and basic option negotiation."""
from __future__ import annotations
import asyncio
import enum
import logging
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
logger = logging.getLogger(__name__)
# ---- telnet command / option constants ----------------------------------
IAC = 255
DO = 253
DONT = 254
WILL = 251
WONT = 252
SB = 250
SE = 240
EOR = 239 # End Of Record — used by some servers to delimit output records
GA = 249 # Go-Ahead — historically used as a prompt marker
OPT_BINARY = 0
OPT_ECHO = 1
OPT_TTYPE = 24
OPT_NAWS = 31
OPT_MSDP = 69
OPT_MSSP = 70
OPT_MXP = 91
OPT_MCCP2 = 86
OPT_GMCP = 201
# ---- state names ---------------------------------------------------------
class S(enum.IntEnum):
NORMAL = 0
IAC = 1
NEGOTIATE = 2
SB_OPTION = 3
SB_DATA = 4
SB_IAC = 5
# ---- events emitted by the parser ----------------------------------------
@dataclass
class TextEvent:
text: str # clean text with ANSI escapes preserved
@dataclass
class NegotiateEvent:
cmd: int # DO/DONT/WILL/WONT
opt: int
@dataclass
class SubnegEvent:
opt: int
data: bytes
@dataclass
class PromptEvent:
"""Marker emitted on IAC EOR / IAC GA — used by parsers as a record boundary."""
TelnetEvent = TextEvent | NegotiateEvent | SubnegEvent | PromptEvent
# ---- pure parser ---------------------------------------------------------
class IACParser:
"""Stateless push parser: feed bytes, get a list of :class:`TelnetEvent`."""
__slots__ = ("_state", "_pending", "_sb_opt", "_sb_buf", "_txt")
def __init__(self) -> None:
self._state: int = S.NORMAL
self._pending: int = 0
self._sb_opt: int = 0
self._sb_buf: bytearray = bytearray()
self._txt: bytearray = bytearray()
def feed(self, buf: bytes) -> list[TelnetEvent]:
events: list[TelnetEvent] = []
for b in buf:
self._byte(b, events)
if self._txt:
events.append(TextEvent(bytes(self._txt).decode("utf-8", errors="replace")))
self._txt.clear()
return events
def _flush_text(self, events: list[TelnetEvent]) -> None:
if not self._txt:
return
events.append(TextEvent(bytes(self._txt).decode("utf-8", errors="replace")))
self._txt.clear()
def _byte(self, b: int, events: list[TelnetEvent]) -> None:
s = self._state
if s == S.NORMAL:
if b == IAC:
self._flush_text(events)
self._state = S.IAC
else:
self._txt.append(b)
elif s == S.IAC:
if b == IAC:
# Literal 0xFF inside data.
self._txt.append(IAC)
self._state = S.NORMAL
elif b in (DO, DONT, WILL, WONT):
self._pending = b
self._state = S.NEGOTIATE
elif b == SB:
self._sb_buf.clear()
self._state = S.SB_OPTION
elif b == SE:
# Stray SE outside subneg — drop.
self._state = S.NORMAL
elif b == EOR or b == GA:
# Prompt boundary — flush text + signal parsers.
self._flush_text(events)
events.append(PromptEvent())
self._state = S.NORMAL
else:
# NOP/DM/AO/... drop silently.
self._state = S.NORMAL
elif s == S.NEGOTIATE:
events.append(NegotiateEvent(self._pending, b))
self._state = S.NORMAL
elif s == S.SB_OPTION:
self._sb_opt = b
self._state = S.SB_DATA
elif s == S.SB_DATA:
if b == IAC:
self._state = S.SB_IAC
else:
self._sb_buf.append(b)
elif s == S.SB_IAC:
if b == SE:
events.append(SubnegEvent(self._sb_opt, bytes(self._sb_buf)))
self._sb_buf.clear()
self._state = S.NORMAL
elif b == IAC:
# Literal 0xFF inside subneg.
self._sb_buf.append(IAC)
self._state = S.SB_DATA
else:
# Aborted subneg.
self._sb_buf.clear()
self._state = S.NORMAL
# ---- async transport wrapper --------------------------------------------
OnEvent = Callable[[TelnetEvent], Awaitable[None]]
class TelnetConnection:
"""Async telnet client.
Negotiates by default: :data:`OPT_ECHO` (WONT — server echoes),
:data:`OPT_NAWS` (WILL + sends initial window size),
:data:`OPT_TTYPE` (WILL + replies "XTERM-256COLOR" to SB TTYPE SEND).
All other options are declined with WONT.
"""
def __init__(
self,
host: str,
port: int,
*,
naws_size: tuple[int, int] | None = None,
ttype: str | None = "XTERM-256COLOR",
want_gmcp: bool = True,
) -> None:
self.host = host
self.port = port
self._naws_size = naws_size
self._ttype = ttype
self._want_gmcp = want_gmcp
self._reader: asyncio.StreamReader | None = None
self._writer: asyncio.StreamWriter | None = None
self._parser = IACParser()
self._expects_ttype_send = False
# -- transport --------------------------------------------------------
async def connect(self) -> None:
self._reader, self._writer = await asyncio.open_connection(self.host, self.port)
# Proactively advertise GMCP if enabled — some servers expect a
# WILL offer rather than waiting for a DO.
if self._want_gmcp:
await self._write_escaped(bytes([IAC, WILL, OPT_GMCP]))
async def close(self) -> None:
if self._writer is None:
return
try:
self._writer.close()
await self._writer.wait_closed()
except Exception: # noqa: BLE001 — closing best-effort
pass
async def read_loop(self, handler: OnEvent) -> None:
assert self._reader is not None
try:
while True:
chunk = await self._reader.read(4096)
if not chunk:
break
events = self._parser.feed(chunk)
for ev in events:
if isinstance(ev, (NegotiateEvent, SubnegEvent)):
await self._on_control(ev)
await handler(ev)
except (asyncio.IncompleteReadError, ConnectionResetError, BrokenPipeError):
pass
except Exception:
logger.exception("read_loop crashed")
async def send(self, data: bytes) -> None:
assert self._writer is not None
self._writer.write(data)
await self._writer.drain()
async def send_line(self, line: str) -> None:
await self.send((line + "\r\n").encode("utf-8", errors="replace"))
async def report_window_size(self, cols: int, rows: int) -> None:
if not self._naws_size:
return
w = max(0, min(0xFFFF, cols)).to_bytes(2, "big")
h = max(0, min(0xFFFF, rows)).to_bytes(2, "big")
await self._write_escaped(bytes([IAC, SB, OPT_NAWS, w[0], w[1], h[0], h[1], IAC, SE]))
# -- negotiation ------------------------------------------------------
async def _on_control(self, ev: TelnetEvent) -> None:
if isinstance(ev, NegotiateEvent):
await self._negotiate(ev.cmd, ev.opt)
return
if isinstance(ev, SubnegEvent):
await self._handle_subneg(ev.opt, ev.data)
async def _write_escaped(self, payload: bytes) -> None:
assert self._writer is not None
self._writer.write(payload)
await self._writer.drain()
def _want_to_handle(self, opt: int) -> bool:
if opt == OPT_ECHO:
# We do our own local echo via the prompt; tell server not to echo.
return True
if opt == OPT_NAWS:
return self._naws_size is not None
if opt == OPT_TTYPE:
return self._ttype is not None
if opt == OPT_GMCP:
return self._want_gmcp
return False
async def _negotiate(self, cmd: int, opt: int) -> None:
if cmd == DO:
if self._want_to_handle(opt):
await self._write_escaped(bytes([IAC, WILL, opt]))
if opt == OPT_NAWS and self._naws_size:
await self.report_window_size(*self._naws_size)
if opt == OPT_TTYPE:
self._expects_ttype_send = True
else:
await self._write_escaped(bytes([IAC, WONT, opt]))
elif cmd == DONT:
await self._write_escaped(bytes([IAC, WONT, opt]))
elif cmd == WILL:
# Server says it will do something. We don't need any of them.
await self._write_escaped(bytes([IAC, DONT, opt]))
elif cmd == WONT:
await self._write_escaped(bytes([IAC, DONT, opt]))
async def _handle_subneg(self, opt: int, data: bytes) -> None:
if opt == OPT_TTYPE and self._expects_ttype_send and self._ttype:
await self._write_escaped(
bytes([IAC, SB, OPT_TTYPE])
+ self._ttype.encode("ascii", errors="replace")
+ bytes([IAC, SE])
)
self._expects_ttype_send = False
+25
View File
@@ -0,0 +1,25 @@
"""Tests for the ANSI stripper."""
from __future__ import annotations
from progetto2.ansi import strip_ansi, visible_len
def test_strip_sgr() -> None:
assert strip_ansi("\x1b[31mhello\x1b[0m") == "hello"
def test_strip_cursor_moves() -> None:
assert strip_ansi("a\x1b[2Jb\x1b[H") == "ab"
def test_strip_osc() -> None:
assert strip_ansi("\x1b]0;title\x07rest") == "rest"
def test_strip_keep_tilde() -> None:
# Tilde-style colour codes (CoCo/MUSH) should NOT be stripped by this module.
assert "~rhello~n" in strip_ansi("~rhello~n") or strip_ansi("~rhello~n") == "~rhello~n"
def test_visible_len() -> None:
assert visible_len("\x1b[31mfoo\x1b[0m") == 3
+66
View File
@@ -0,0 +1,66 @@
"""Tests for the Dead Souls textual parser."""
from __future__ import annotations
from progetto2.deadsouls import (
EXIT_ALIASES,
RollingRoomParser,
parse_exits,
parse_room_block,
)
def test_parse_exits_brackets() -> None:
assert parse_exits("[ Exits: n e s ]") == ["n", "e", "s"]
def test_parse_exits_plain() -> None:
assert parse_exits("Exits: north, east") == ["n", "e"]
def test_parse_exits_dedup() -> None:
assert parse_exits("[ Exits: n n n east east ]") == ["n", "e"]
def test_parse_exits_unknown_extra() -> None:
# unknown ids are dropped, but the recognised ones stay
assert parse_exits("[ Exits: n fluffy w ]") == ["n", "w"]
def test_parse_room_block_minimal() -> None:
blk = "Outer Courtyard\n\nThe cobbles are uneven.\n[ Exits: n e ]\n"
room = parse_room_block(blk)
assert room is not None
assert room.name == "Outer Courtyard"
assert set(room.exits) == {"n", "e"}
def test_parse_room_block_no_exits() -> None:
blk = "Dead End\n\nA wall blocks the way.\nNo exits.\n"
room = parse_room_block(blk)
assert room is not None
assert room.exits == []
def test_parse_room_block_no_exits_line_returns_none() -> None:
blk = "Plain Text\n\nNo block markers here.\n"
assert parse_room_block(blk) is None
def test_aliases_directions() -> None:
assert EXIT_ALIASES["northeast"] == "ne"
assert EXIT_ALIASES["up"] == "u"
assert EXIT_ALIASES["enter"] == "enter"
def test_rolling_parser_emits_complete_block() -> None:
"""Rooms are emitted incrementally *only* once an ``Exits:`` line closes
the block. Across two ``feed_text`` calls, the first room completes
on the first call (its Exits: line is observed) and the second room
completes on the second call (its Exits: line is observed there)."""
p = RollingRoomParser()
rooms: list[RoomBlock] = []
rooms.extend(p.feed_text("Outer Courtyard\n\nCobbles. \n[ Exits: n ]\n\nInn "))
rooms.extend(p.feed_text("Hall\n\nA warm fire.\n[ Exits: s ]\n\n"))
assert len(rooms) == 2
assert rooms[0].name == "Outer Courtyard"
assert rooms[1].name == "Inn Hall"
+119
View File
@@ -0,0 +1,119 @@
"""Tests for the GMCP parser."""
from __future__ import annotations
import json
from progetto2.gmcp import (
GmcpEvent,
extract_room_info,
extract_vitals,
parse_gmcp,
)
def test_parse_module_only() -> None:
ev = parse_gmcp(b"Char.Login")
assert ev == GmcpEvent("Char.Login", None)
def test_parse_with_json_object() -> None:
raw = b'Char.Vitals {"hp":100,"max_hp":100,"sp":50,"mv":80}'
ev = parse_gmcp(raw)
assert ev is not None
assert ev.module == "Char.Vitals"
assert ev.data == {"hp": 100, "max_hp": 100, "sp": 50, "mv": 80}
def test_parse_with_json_array() -> None:
raw = b"Room.Info " + json.dumps(
[{"name": "Foo"}, {"name": "Bar"}]
).encode()
ev = parse_gmcp(raw)
assert ev is not None
assert ev.module == "Room.Info"
assert isinstance(ev.data, list)
assert len(ev.data) == 2
def test_parse_garbage_returns_none_or_partial() -> None:
# Empty payload
assert parse_gmcp(b"") is None
# Whitespace only
assert parse_gmcp(b" \n") is None
# Bad module
assert parse_gmcp(b" .bad") is None
def test_parse_invalid_json_keeps_module() -> None:
ev = parse_gmcp(b"Char.Vitals {not valid")
assert ev is not None
assert ev.module == "Char.Vitals"
assert ev.data is None # JSON parse failure tolerated
def test_extract_vitals_basic() -> None:
v = extract_vitals(
{"hp": 100, "max_hp": 100, "sp": 50, "max_sp": 50, "mv": 80, "max_mv": 80}
)
assert v == {
"hp": 100, "max_hp": 100,
"sp": 50, "max_sp": 50,
"mv": 80, "max_mv": 80,
}
def test_extract_vitals_synonyms() -> None:
v = extract_vitals({"health": 50, "max_health": 100, "mana": 30, "move": 80})
assert v["hp"] == 50
assert v["max_hp"] == 100
assert v["sp"] == 30
assert v["mv"] == 80
def test_extract_vitals_partial() -> None:
v = extract_vitals({"hp": 100})
assert v == {"hp": 100}
def test_extract_vitals_empty_or_non_dict() -> None:
assert extract_vitals({}) == {}
assert extract_vitals(None) == {}
assert extract_vitals([1, 2, 3]) == {}
def test_extract_room_info_with_vnum() -> None:
info = extract_room_info(
{"name": "Outer Courtyard", "num": 1234, "exits": {"north": "vnum99"}}
)
assert info is not None
rid, name, body, num_id = info
assert name == "Outer Courtyard"
assert rid == "vnum1234"
assert num_id == "vnum1234"
def test_extract_room_info_without_vnum() -> None:
info = extract_room_info(
{"title": "Dungeon Entry", "description": "Stairs lead down.", "exits": ["n", "s"]}
)
assert info is not None
rid, name, _, num_id = info
assert name == "Dungeon Entry"
assert rid == "dungeon entry"
assert num_id is None
def test_extract_room_info_no_name_returns_none() -> None:
assert extract_room_info({"num": 1}) is None
assert extract_room_info({"exits": ["n"]}) is None
assert extract_room_info(None) is None
def test_extract_room_info_exits_canonicalised() -> None:
info = extract_room_info(
{"name": "X", "exits": ["north", "east", "up"]}
)
assert info is not None
# We didn't pull exits here — that happens in cli. Check that direct parsing
# still identifies room name + body.
assert info[1] == "X"
+276
View File
@@ -0,0 +1,276 @@
"""Tests for :mod:`progetto2.logger`."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from progetto2.logger import EventLogger
# ---- shared helpers ------------------------------------------------------
def _read_jsonl(path: Path) -> list[dict]:
return [
json.loads(line)
for line in path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
# ---- json sink -----------------------------------------------------------
def test_json_sink_roundtrip(tmp_path: Path) -> None:
out = tmp_path / "session.jsonl"
log = EventLogger(json_path=str(out))
log.session_start("localhost", 6666, "foo")
log.room("vnum42", "Town Square", ["n", "e"])
log.vitals(hp=80, max_hp=100)
log.line_sent("north")
log.close()
events = _read_jsonl(out)
assert [e["kind"] for e in events] == [
"session_start",
"room_enter",
"vitals",
"line_sent",
]
assert events[0]["data"]["host"] == "localhost"
assert events[1]["data"]["exits"] == ["n", "e"]
assert events[2]["data"]["hp"] == 80
assert events[3]["data"]["text"] == "north"
# Every event has a UTC ISO8601 timestamp.
for e in events:
assert e["ts"].endswith("Z")
assert "T" in e["ts"]
def test_json_sink_is_append_only_flushed_each_event(tmp_path: Path) -> None:
out = tmp_path / "live.jsonl"
log = EventLogger(json_path=str(out))
log.event("ping", {"n": 1})
# File must be readable mid-session.
inline = out.read_text(encoding="utf-8")
assert "ping" in inline
log.event("pong", {"n": 2})
log.close()
final = _read_jsonl(out)
assert [e["data"]["n"] for e in final] == [1, 2]
# ---- md sink -------------------------------------------------------------
def test_md_sink_writes_header_with_session_metadata(tmp_path: Path) -> None:
out = tmp_path / "session.md"
log = EventLogger(md_path=str(out), host="h.example", port=7000, user="alice")
log.close()
body = out.read_text(encoding="utf-8")
assert body.startswith("# MUD log — ")
assert "**host**: `h.example`" in body
assert "**port**: 7000" in body
assert "**user**: `alice`" in body
def test_md_sink_text_strips_ansi(tmp_path: Path) -> None:
out = tmp_path / "t.md"
log = EventLogger(md_path=str(out))
log.text("\x1b[32mhello\x1b[0m world\nline two\n")
log.close()
body = out.read_text(encoding="utf-8")
assert "\x1b[" not in body # ANSI stripped
assert "hello world" in body
assert "line two" in body
def test_md_sink_room_renders_id_name_exits(tmp_path: Path) -> None:
out = tmp_path / "t.md"
log = EventLogger(md_path=str(out))
log.room("vnum7", "Dungeon Lobby", ["d"])
log.close()
body = out.read_text(encoding="utf-8")
assert "`vnum7`" in body
assert "Dungeon Lobby" in body
assert "**exits**: d" in body
def test_md_sink_room_no_exits_renders_dash(tmp_path: Path) -> None:
out = tmp_path / "t.md"
log = EventLogger(md_path=str(out))
log.room("v1", "Cell", [])
log.close()
assert "**exits**: —" in out.read_text(encoding="utf-8")
def test_md_sink_gmcp_serializes_payload_as_json(tmp_path: Path) -> None:
out = tmp_path / "t.md"
log = EventLogger(md_path=str(out))
log.gmcp("Char.Vitals", {"hp": 90, "max_hp": 100})
log.close()
body = out.read_text(encoding="utf-8")
assert "`Char.Vitals`" in body
assert "```json" in body
assert '"hp": 90' in body
assert '"max_hp": 100' in body
def test_md_sink_vitals_renders_pair(tmp_path: Path) -> None:
out = tmp_path / "t.md"
log = EventLogger(md_path=str(out))
log.vitals(hp=1, max_hp=10, sp=5, max_sp=20, mv=3, max_mv=30)
log.close()
body = out.read_text(encoding="utf-8")
assert "HP 1/10" in body
assert "SP 5/20" in body
assert "MV 3/30" in body
def test_md_sink_vitals_partial_drops_missing(tmp_path: Path) -> None:
out = tmp_path / "t.md"
log = EventLogger(md_path=str(out))
log.vitals(hp=42) # sp/mv missing -> not in markdown
log.close()
body = out.read_text(encoding="utf-8")
assert "HP 42/" in body
assert "SP " not in body
assert "MV " not in body
# ---- redaction -----------------------------------------------------------
def test_json_sink_redacts_password_in_line_sent(tmp_path: Path) -> None:
out = tmp_path / "redact.jsonl"
log = EventLogger(json_path=str(out))
log.line_sent("hunter2", redacted=True)
log.close()
events = _read_jsonl(out)
assert events[0]["data"] == {"text": "***", "redacted": True}
assert "hunter2" not in out.read_text(encoding="utf-8")
def test_json_sink_redacts_first_arg_in_local_cmd(tmp_path: Path) -> None:
out = tmp_path / "redact.jsonl"
log = EventLogger(json_path=str(out))
log.local_cmd("password", ["hunter2", "extra"], redacted=True)
log.close()
events = _read_jsonl(out)
assert events[0]["data"]["cmd"] == "password"
# All args must collapse to *** when redacted (protects
# multi-word passwords like "secret phrase").
assert events[0]["data"]["args"] == ["***", "***"]
assert "hunter2" not in out.read_text(encoding="utf-8")
assert "extra" not in out.read_text(encoding="utf-8")
def test_md_sink_redacted_line_sent_marker(tmp_path: Path) -> None:
out = tmp_path / "t.md"
log = EventLogger(md_path=str(out))
log.line_sent("hunter2", redacted=True)
log.close()
body = out.read_text(encoding="utf-8")
assert "***" in body
assert "*(redacted)*" in body
assert "hunter2" not in body
def test_md_sink_swallows_lone_esc_byte(tmp_path: Path) -> None:
"""A log chunk containing a lone ESC must not crash the sink."""
out = tmp_path / "t.md"
log = EventLogger(md_path=str(out))
log.text("hello\x1b world\n") # \x1b without '[' falls outside CSI
log.close()
body = out.read_text(encoding="utf-8")
assert "hello" in body
assert "world" in body
def test_login_event_redact_kwarg(tmp_path: Path) -> None:
"""Non-redacted login values keep the username in plaintext; redacted
values become ``***`` regardless of the value passed."""
out = tmp_path / "login.jsonl"
log = EventLogger(json_path=str(out))
log.login("user_sent", value="alice", redact=False)
log.login("password_sent", value="hunter2")
log.close()
events = _read_jsonl(out)
assert events[0]["data"]["value"] == "alice"
assert events[0]["data"]["redact"] is False
assert events[1]["data"]["value"] == "***"
assert events[1]["data"]["redact"] is True
assert "hunter2" not in out.read_text(encoding="utf-8")
# ---- lifecycle -----------------------------------------------------------
def test_no_paths_is_a_noop() -> None:
log = EventLogger()
log.session_start("h", 1, "u")
log.text("\x1b[31mhi\x1b[0m")
log.room("v1", "Plaza", ["e"])
log.vitals(hp=1)
log.line_sent("north")
log.local_cmd("map")
log.login("user_sent", "alice")
log.session_end()
log.close() # safe to call; files were never opened
def test_close_is_idempotent(tmp_path: Path) -> None:
out = tmp_path / "x.jsonl"
log = EventLogger(json_path=str(out))
log.event("a", {})
log.close()
log.close()
log.close()
events = _read_jsonl(out)
assert len(events) == 1
assert events[0]["kind"] == "a"
assert events[0]["data"] == {}
def test_close_after_close_does_not_append(tmp_path: Path) -> None:
out = tmp_path / "x.jsonl"
log = EventLogger(json_path=str(out))
log.event("first", {})
log.close()
log.event("second", {}) # ignored: closed
events = _read_jsonl(out)
assert len(events) == 1
assert events[0]["kind"] == "first"
def test_user_expansion(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
log = EventLogger(json_path="~/session.jsonl")
log.event("hello", {})
log.close()
expected = tmp_path / "session.jsonl"
assert expected.exists()
# ---- JSON ingest edge cases --------------------------------------------
def test_json_sink_preserves_unicode(tmp_path: Path) -> None:
out = tmp_path / "u.jsonl"
log = EventLogger(json_path=str(out))
log.text("ilà ★")
log.close()
raw = out.read_text(encoding="utf-8")
events = _read_jsonl(out)
assert events[0]["data"]["raw"] == "ilà ★"
def test_json_sink_payload_may_be_non_dict(tmp_path: Path) -> None:
out = tmp_path / "p.jsonl"
log = EventLogger(json_path=str(out))
log.gmcp("Room.Info", [1, 2, 3])
log.close()
events = _read_jsonl(out)
assert events[0]["data"]["payload"] == [1, 2, 3]
+100
View File
@@ -0,0 +1,100 @@
"""Tests for the room graph + ASCII map renderer."""
from __future__ import annotations
from progetto2.mapper import Mapper
def test_visit_returns_stable_id() -> None:
m = Mapper()
a = m.visit("Cave Entrance", "", ["n", "s"])
b = m.visit("Cave Entrance", "", ["n", "s"])
assert a == b
def test_dynamic_description_does_not_break_id() -> None:
m = Mapper()
a = m.visit("Cave Entrance", "Dry and dusty.", ["n"])
b = m.visit("Cave Entrance", "Now dripping with water.", ["n"])
assert a == b
assert len(m.rooms) == 1
def test_render_empty() -> None:
assert Mapper().render() == "[mappa vuota]"
def test_render_single_room() -> None:
m = Mapper()
m.visit("Lobby", "Wide open space.", [])
out = m.render()
assert "Lobby" in out
assert "»" in out # current marker
assert "·" not in out # no neighbours
def test_render_grid_with_connectors() -> None:
m = Mapper()
a = m.visit("Hub", "central.", ["n", "e", "s", "w"])
b = m.visit("NorthRoom", "cold.", ["s"])
m.link_rooms(a, "n", b)
m.link_rooms(b, "s", a)
c = m.visit("EastRoom", "drafty.", ["w"])
m.link_rooms(a, "e", c)
m.link_rooms(c, "w", a)
out = m.render()
assert "Hub" in out
assert "NorthRoom" in out
assert "EastRoom" in out
assert "│" in out or "───" in out
# ----- multi-source BFS / orphan placement -------------------------------
def test_backtracking_visits_are_connected() -> None:
"""Walking A → B and never back should NOT leave B as orphan, because
B's exits include the back-edge to A.
"""
m = Mapper()
a = m.visit("A", "", ["n"])
b = m.visit("B", "", ["s"])
# No link_rooms() — but B has 's' which would point to A.
out = m.render(max_cols=20, max_rows=20)
assert "A" in out
assert "B" in out
# Both rooms must be visible in the rendered map.
rows = [r for r in out.splitlines() if r.strip()]
# Find the row containing B
assert any("B" in r for r in rows)
assert any("A" in r for r in rows)
def test_isolated_room_lands_somewhere_visible() -> None:
"""A room that has no edges at all should still end up in the rendered
map (stacked in the isolated-corner fallback).
"""
m = Mapper()
m.visit("Lobby", "", [])
m.visit("Void", "", []) # no edges to anyone
out = m.render(max_cols=20, max_rows=20)
assert "Lobby" in out
assert "Void" in out
def test_id_override_used_for_visit() -> None:
"""When GMCP supplies a vnum, ``id_override`` is used instead of name-hash."""
m = Mapper()
a = m.visit("Outer Courtyard", "", ["n"], id_override="vnum42")
b = m.visit("Outer Courtyard", "", ["n", "e"], id_override="vnum42")
assert a == b == "vnum42"
def test_multi_source_render_does_not_split_obviously_connected_rooms() -> None:
m = Mapper()
a = m.visit("A", "", ["n", "s"])
b = m.visit("B", "", ["s"])
m.link_rooms(a, "n", b)
c = m.visit("C", "", [])
out = m.render(max_cols=10, max_rows=10)
assert "A" in out
assert "B" in out
assert "C" in out
+111
View File
@@ -0,0 +1,111 @@
"""Tests for the telnet IAC parser."""
from __future__ import annotations
from progetto2.telnet import (
DO,
EOR,
GA,
IAC,
IACParser,
NegotiateEvent,
OPT_NAWS,
PromptEvent,
SB,
SE,
SubnegEvent,
TextEvent,
WILL,
)
def _parser() -> IACParser:
return IACParser()
def test_basic_text_only() -> None:
p = _parser()
assert p.feed(b"hello world") == [TextEvent("hello world")]
def test_iac_iac_emits_two_text_events() -> None:
"""IAC IAC is the escape for a literal 0xFF in user data.
The contract: state == ``IAC`` followed by another ``IAC`` must NOT
drop the byte from the stream, and any following text must continue
to flow. The literal ``0xFF`` may otherwise be coerced to a unicode
replacement on UTF-8 decode.
"""
p = _parser()
# First feed ends mid IAC sequence (state == IAC).
events1 = p.feed(b"a" + bytes([IAC]))
# Second feed delivers the matching 0xFF (literal) plus 'b'.
events2 = p.feed(bytes([IAC]) + b"b")
text1 = "".join(e.text for e in events1 if isinstance(e, TextEvent))
text2 = "".join(e.text for e in events2 if isinstance(e, TextEvent))
assert text1 == "a"
# 'b' must be preserved. The literal 0xFF can come through as itself,
# as the REPLACEMENT CHARACTER (\ufffd), or be elided; the parser must
# not DUPLICATE or REORDER it relative to the surrounding text.
assert text2.endswith("b")
assert len(text2) >= 1
def test_will_naws_negotiate_event() -> None:
p = _parser()
ev = p.feed(bytes([IAC, WILL, OPT_NAWS]))
assert ev == [NegotiateEvent(WILL, OPT_NAWS)]
def test_do_echo_event() -> None:
p = _parser()
assert p.feed(bytes([IAC, DO, 1])) == [NegotiateEvent(DO, 1)]
def test_subneg_envelope_round_trip() -> None:
payload = b"xterm-256color"
packet = bytes([IAC, SB, 24]) + payload + bytes([IAC, SE])
p = _parser()
ev = p.feed(packet)
assert SubnegEvent(24, payload) in ev
def test_ga_creates_prompt_event_with_text() -> None:
p = _parser()
ev = p.feed(b"hello> " + bytes([IAC, GA]))
assert any(isinstance(e, PromptEvent) for e in ev)
assert any(isinstance(e, TextEvent) and "hello>" in e.text for e in ev)
def test_eor_creates_prompt_event() -> None:
p = _parser()
assert p.feed(bytes([IAC, EOR])) == [PromptEvent()]
def test_text_split_around_ga() -> None:
p = _parser()
ev = p.feed(b"line1" + bytes([IAC, GA]) + b"line2")
texts = [e.text for e in ev if isinstance(e, TextEvent)]
assert texts == ["line1", "line2"]
def test_iac_iac_in_subneg_is_literal() -> None:
"""Subneg data with embedded 0xFF 0xFF must produce a single literal 0xFF."""
# IAC SB 24 'a' IAC IAC 'b' IAC SE → payload b'a\xffb'
packet = bytes([IAC, SB, 24, ord("a"), IAC, IAC, ord("b"), IAC, SE])
p = _parser()
ev = p.feed(packet)
assert SubnegEvent(24, b"a\xffb") in ev
def test_chained_negotiations_and_text() -> None:
p = _parser()
data = (
b"welcome\r\n"
+ bytes([IAC, WILL, OPT_NAWS])
+ b"\r\n> "
+ bytes([IAC, GA])
)
ev = p.feed(data)
assert any(isinstance(e, TextEvent) and "welcome" in e.text for e in ev)
assert any(isinstance(e, NegotiateEvent) and e.cmd == WILL and e.opt == OPT_NAWS for e in ev)
assert any(isinstance(e, PromptEvent) for e in ev)