5 - Python
fft-web/fft_web.py
fft-web/fft_web.py
#!/usr/bin/env python3
# fft_web.py — 2 FFT (MS + BTS) sur UNE page, serveur web autonome (port 8081).
#
# Reprend la logique de tools/fft.sh / tools/fft2.sh (PSD Welch sur l'I/Q fc32) mais :
# - NATIF : lit directement les sources de l'hôte, pas de docker exec.
# - HEADLESS : pas de matplotlib/X. numpy calcule la PSD → JSON → le navigateur dessine (canvas).
# - LIVE FIFO : le spectre BTS lit /tmp/iq_fft.fifo EN CONTINU (le tap FFT
# dédié du relay DL osmo-trx) au lieu du ring record.cfile (128 Mo) qui
# FIGEAIT une fois plein. La FIFO est ouverte O_RDWR|O_NONBLOCK (comme
# grgsm_fft_live.py) : pas d'EOF, le write non-bloquant du relay trouve
# toujours un lecteur, et on ne perturbe pas la boucle relay / le camping.
#
# - MS = iq_fft_ms.fifo (montant, ecrit par pont.py ; LIVE)
# - BTS = iq_fft.fifo (DL relay osmo-trx, LIVE)
# Chaque source accepte indifféremment un .cfile (lecture tail) OU un .fifo
# (lecture live) : auto-détecté. Override par env CFILE_MS / CFILE_BTS.
#
# Lancement : python3 /opt/GSM/osmo-nitb-for-calypso/fft-web/fft_web.py
# FFT_WEB_PORT=8081 RATE=1083333 NSAMP=262144 NFFT=4096 ... (overridables par env)
import os, json, stat, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
import numpy as np
# --- racine de l'installation, resolue sans chemin en dur --------------------
# GSM_ROOT si l'environnement le pose (c'est le cas quand on passe par run.sh),
# sinon le parent du depot : les deux depots vivent cote a cote.
import os as _os
GSM_ROOT = _os.environ.get(
"GSM_ROOT",
_os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))))
PORT = int(os.environ.get('FFT_WEB_PORT', '8081'))
RATE = float(os.environ.get('RATE', '1083333')) # 4 SPS natif = 26e6/24 (= Fs du relay/FIFO)
NSAMP = int(os.environ.get('NSAMP', '262144')) # complex samples gardés (tail / fenêtre live)
NFFT = int(os.environ.get('NFFT', '4096')) # résolution FFT (segment Welch)
# Cadence d'affichage : 40 ms = 25 img/s. Le navigateur poll /psd à cet intervalle
# (2 fetch ms+bts par tick → ~50 req/s). On borne le nb de segments Welch par
# requête (NSEG_MAX) pour que le calcul tienne la cadence (NFFT*NSEG_MAX FFT/req).
REFRESH_MS = int(os.environ.get('REFRESH_MS', '40'))
NSEG_MAX = int(os.environ.get('NSEG_MAX', '16'))
SRC = {
# [2026-08-29] Le montant vient de pont.py (PONT_AIRREC_UL_FIFO), pas du
# cfile DSP : en mode pont, /dev/shm/dsp_iq.cfile n'a aucun producteur et
# le spectre MS restait vide - une seule FFT vivante sur les deux.
'ms': {'path': os.environ.get('CFILE_MS', '/tmp/iq_fft_ms.fifo'),
'arfcn': os.environ.get('ARFCN_MS', '514'), 'label': 'MS — montant (iq_fft_ms.fifo)'},
'bts': {'path': os.environ.get('CFILE_BTS', '/tmp/iq_fft.fifo'),
'arfcn': os.environ.get('ARFCN_BTS', '514'), 'label': 'BTS — DL relay LIVE (iq_fft.fifo)'},
}
# ── État live par source FIFO : fd O_RDWR|O_NONBLOCK + buffer roulant ────────
_state = {} # src -> {'fd':int, 'buf':bytearray}
_locks = {k: threading.Lock() for k in SRC}
MAXB = NSAMP * 8 # octets gardés (complex64 = 8 o/échantillon)
def _is_fifo(path):
try:
return stat.S_ISFIFO(os.stat(path).st_mode)
except OSError:
return path.endswith('.fifo') # pas encore créé : on se fie au suffixe
def read_live_iq(src, path):
"""Draine la FIFO en non-bloquant et renvoie les NSAMP derniers échantillons
(le plus frais). O_RDWR => jamais d'EOF, jamais de blocage du relay."""
st = _state.get(src)
if st is None:
if not os.path.exists(path):
os.mkfifo(path, 0o666)
fd = os.open(path, os.O_RDWR | os.O_NONBLOCK)
st = {'fd': fd, 'buf': bytearray()}
_state[src] = st
fd, buf = st['fd'], st['buf']
while True:
try:
c = os.read(fd, 1 << 16)
except BlockingIOError:
break
if not c:
break
buf += c
if len(buf) > MAXB:
del buf[:-MAXB] # borne la RAM, garde le frais
if len(buf) > MAXB:
del buf[:-MAXB]
n = (len(buf) // 8) * 8
if n < NFFT * 8:
return np.array([], dtype=np.complex64)
return np.frombuffer(bytes(buf[-n:]), dtype=np.complex64)
def read_tail_iq(path, nsamp):
sz = os.path.getsize(path)
nbytes = min(sz - (sz % 8), nsamp * 8) # complex64 = 8 octets/échantillon
with open(path, 'rb') as f:
if sz > nbytes:
f.seek(sz - nbytes)
raw = f.read(nbytes)
return np.frombuffer(raw, dtype=np.complex64)
def welch_psd(iq, nfft, rate):
if iq.size < nfft:
return None, None
win = np.hanning(nfft).astype(np.float32)
nseg = min(iq.size // nfft, NSEG_MAX) # borne le coût pour tenir 25 img/s
iq = iq[-nseg*nfft:] # garde les segments les plus frais
acc = np.zeros(nfft, dtype=np.float64)
for i in range(nseg):
seg = iq[i*nfft:(i+1)*nfft] * win
acc += np.abs(np.fft.fftshift(np.fft.fft(seg)))**2
acc /= nseg
psd_db = 10.0*np.log10(acc + 1e-12)
freqs = np.fft.fftshift(np.fft.fftfreq(nfft, 1.0/rate)) / 1e3 # kHz
return freqs, psd_db
def psd_json(src):
s = SRC[src]
path = s['path']
try:
with _locks[src]:
iq = read_live_iq(src, path) if _is_fifo(path) else read_tail_iq(path, NSAMP)
f, p = welch_psd(iq, NFFT, RATE)
if f is None:
return {'error': "flux pas encore prêt (pas assez d'échantillons)", 'label': s['label']}
step = max(1, len(f)//1024) # sous-échantillonne pour le transport
return {'label': s['label'], 'arfcn': s['arfcn'], 'rate': RATE,
'freqs': [round(x, 1) for x in f[::step].tolist()],
'psd': [round(x, 2) for x in p[::step].tolist()]}
except FileNotFoundError:
return {'error': 'source absente (%s) — lance la stack' % path, 'label': s['label']}
except Exception as e:
return {'error': str(e), 'label': s['label']}
PAGE = """<!doctype html><html><head><meta charset=utf-8><title>Calypso FFT — MS & BTS</title>
<style>body{background:#0b0f14;color:#cdd6e0;font-family:monospace;margin:0;padding:12px}
h1{font-size:15px;color:#2aa198;margin:0 0 10px} .row{display:flex;gap:12px;flex-wrap:wrap}
.card{flex:1;min-width:380px;background:#0e151c;border:1px solid #1d2a33;border-radius:8px;padding:8px}
.t{font-size:12px;color:#b58900;margin-bottom:4px} canvas{width:100%;background:#060a0e;border-radius:4px;display:block}
.psd{height:200px} .wf{height:150px;margin-top:6px;image-rendering:pixelated}
.err{color:#dc322f;font-size:12px;min-height:14px}</style></head><body>
<h1>\U0001F4E1 Calypso I/Q FFT — MS vs BTS <span id=st style="color:#586e75;font-size:11px"></span></h1>
<div class=row>
<div class=card><div class=t id=t-ms>MS</div><canvas id=c-ms class=psd></canvas><canvas id=w-ms class=wf></canvas><div class=err id=e-ms></div></div>
<div class=card><div class=t id=t-bts>BTS</div><canvas id=c-bts class=psd></canvas><canvas id=w-bts class=wf></canvas><div class=err id=e-bts></div></div>
</div>
<script>
function cmap(v){v=Math.max(0,Math.min(1,v));var h=(1-v)*240,l=0.12+0.5*v,a=1*Math.min(l,1-l);
var f=function(n){var k=(n+h/30)%12;return Math.round(255*(l-a*Math.max(-1,Math.min(k-3,9-k,1))));};
return [f(0),f(8),f(4)];}
function drawPsd(src,d){var c=document.getElementById('c-'+src),x=c.getContext('2d');
var W=c.width=c.clientWidth*2,H=c.height=c.clientHeight*2;x.clearRect(0,0,W,H);
var p=d.psd,n=p.length;if(!n)return;var mn=Math.min.apply(null,p),mx=Math.max.apply(null,p),rg=(mx-mn)||1;
x.strokeStyle='#16222b';x.lineWidth=1;for(var g=0;g<=4;g++){var yy=H*g/4;x.beginPath();x.moveTo(0,yy);x.lineTo(W,yy);x.stroke();}
x.beginPath();x.moveTo(W/2,0);x.lineTo(W/2,H);x.strokeStyle='#243845';x.stroke();
x.strokeStyle='#2aa198';x.lineWidth=2;x.beginPath();
for(var i=0;i<n;i++){var xx=W*i/(n-1),yy=H-(p[i]-mn)/rg*H*0.9-H*0.05;i?x.lineTo(xx,yy):x.moveTo(xx,yy);}
x.stroke();}
function drawWf(src,d){var c=document.getElementById('w-'+src),x=c.getContext('2d');
var p=d.psd,n=p.length;if(!n)return;
if(c.width!==n){c.width=n;c.height=160;x.fillStyle='#060a0e';x.fillRect(0,0,n,c.height);}
// [2026-08-09] SENS DE DEFILEMENT REMIS A L'ENDROIT. C'etait drawImage(c,0,-1)
// + putImageData(...,c.height-1) : le plus recent en BAS, l'historique qui
// monte. Toutes les cascades SDR (gqrx, GNU Radio, grgsm) font l'inverse — le
// plus recent EN HAUT, l'historique qui descend — et l'oeil lit le temps de
// haut en bas. On decale donc l'image vers le BAS et on ecrit la ligne neuve
// sur la PREMIERE ligne.
x.drawImage(c,0,1); // scroll vers le bas d'1px
var mn=Math.min.apply(null,p),mx=Math.max.apply(null,p),rg=(mx-mn)||1;
var row=x.createImageData(n,1);
for(var i=0;i<n;i++){var col=cmap((p[i]-mn)/rg),o=i*4;row.data[o]=col[0];row.data[o+1]=col[1];row.data[o+2]=col[2];row.data[o+3]=255;}
x.putImageData(row,0,0);} // nouvelle ligne en haut
function draw(src,d){var e=document.getElementById('e-'+src),tt=document.getElementById('t-'+src);
if(d.error){e.textContent=d.error;tt.textContent=d.label||'';return;} e.textContent='';
tt.textContent=d.label+' • ARFCN '+d.arfcn+' • Fs '+(d.rate/1e6).toFixed(3)+' MHz';
drawPsd(src,d);drawWf(src,d);}
function tick(){['ms','bts'].forEach(function(s){
fetch('/psd?src='+s+'&t='+Date.now()).then(function(r){return r.json();}).then(function(d){draw(s,d);}).catch(function(){});});
document.getElementById('st').textContent='— '+new Date().toLocaleTimeString();}
tick();setInterval(tick,__REFRESH_MS__);
</script></body></html>"""
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def do_GET(self):
u = urlparse(self.path)
if u.path == '/psd':
q = parse_qs(u.query); src = q.get('src', ['ms'])[0]
if src not in SRC:
src = 'ms'
body = json.dumps(psd_json(src)).encode()
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Cache-Control', 'no-store')
self.send_header('Content-Length', str(len(body)))
self.end_headers(); self.wfile.write(body); return
body = PAGE.replace('__REFRESH_MS__', str(REFRESH_MS)).encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Content-Length', str(len(body)))
self.end_headers(); self.wfile.write(body)
if __name__ == '__main__':
print('FFT web on :%d MS=%s BTS=%s' % (PORT, SRC['ms']['path'], SRC['bts']['path']))
ThreadingHTTPServer(('0.0.0.0', PORT), Handler).serve_forever()opt-gsm/grgsm_fft_live.py
opt-gsm/grgsm_fft_live.py
#!/usr/bin/env python3
# grgsm_fft_live.py — FFT EN DIRECT (spectre ASCII). Lit le FIFO live
# (defaut /tmp/iq_asciifft.fifo, fc32 4 SPS) que qemu pousse trame-par-trame,
# OU un cfile fichier (seek-tail) en fallback. Marque le pic FCCH.
#
# FIFO : ouvert O_RDWR|O_NONBLOCK -> open non-bloquant + pas d'EOF (on garde un
# write-end) => le write non-bloquant de qemu trouve toujours un lecteur. A
# chaque tour on DRAINE le backlog et on ne garde que les N derniers samples
# (le plus frais), donc pas de lag qui s'accumule.
import numpy as np, time, os, stat
CFILE=os.environ.get("CFILE","/tmp/iq_asciifft.fifo")
FS=float(os.environ.get("FS","1083333")) # 4 SPS relay (FIFO). cfile BSP 1 SPS=270833
N=int(os.environ.get("NFFT","4096"))
COLS=70
FCCH=1625000.0/24 # +67708 Hz
def _is_fifo(p):
try: return stat.S_ISFIFO(os.stat(p).st_mode)
except OSError: return False
fifo_fd=None
if _is_fifo(CFILE) or CFILE.endswith(".fifo"):
if not os.path.exists(CFILE): os.mkfifo(CFILE,0o666)
fifo_fd=os.open(CFILE, os.O_RDWR|os.O_NONBLOCK)
def _read_iq():
if fifo_fd is not None:
buf=b""
while True:
try: c=os.read(fifo_fd, 1<<16)
except BlockingIOError: break
if not c: break
buf+=c
if len(buf) > N*8*4: buf=buf[-(N*8):] # garde le frais, borne la RAM
buf=buf[-(N*8):]
if len(buf) < N*8: return np.array([],dtype=np.complex64)
return np.frombuffer(buf[:(len(buf)//8)*8], dtype=np.complex64)
try: sz=os.path.getsize(CFILE)
except OSError: sz=0
if sz < N*8: return np.array([],dtype=np.complex64)
with open(CFILE,"rb") as f:
f.seek(max(0,(sz//8-N)*8)); return np.frombuffer(f.read(N*8),dtype=np.complex64)
while True:
iq=_read_iq()
os.system("clear")
print("== FFT LIVE %s (%s, fs=%.0f, N=%d) == FCCH attendu %+.0f Hz"%(
CFILE, "FIFO" if fifo_fd is not None else "file", FS, N, FCCH))
if len(iq) < N:
print("\n pas (encore) de donnees -> qemu ecrit-il le FIFO ? (relance ./run.sh)")
time.sleep(1); continue
sp=np.abs(np.fft.fftshift(np.fft.fft(iq*np.hanning(len(iq)))))
fr=np.fft.fftshift(np.fft.fftfreq(len(iq),1/FS))
pk=fr[np.argmax(sp)]; mag=np.abs(iq)
print(" ampl moy=%.5f max=%.4f | PIC %+.0f Hz | dphi std=%.2f (GMSK~1.57)\n"%(
mag.mean(),mag.max(),pk,np.std(np.angle(iq[1:]*np.conj(iq[:-1])))))
B=24; step=len(sp)//B; spb=np.array([sp[i*step:(i+1)*step].max() for i in range(B)])
frb=np.array([fr[i*step] for i in range(B)]); m=spb.max()+1e-9
for i in range(B):
bar=int(COLS*spb[i]/m); tag=" <-FCCH" if abs(frb[i]-FCCH)<FS/B else ""
print("%+7.0f|%s%s"%(frb[i],"#"*bar,tag))
time.sleep(0.5)opt-gsm/record_drain.py
opt-gsm/record_drain.py
#!/usr/bin/env python3
# record_drain.py — draine le FIFO live iq_record.fifo -> record.cfile en RING
# (128 MB, fseek wrap), HORS du hot-path qemu. C'est l'EXTERNALISATION du ring
# cfile que qemu ecrivait avant dans son hot-path DL (= ce qui causait les
# underruns). Ici qemu ne fait qu'un write() non-bloquant vers le FIFO ; ce
# process (independant) absorbe le flux et l'ecrit sur disque a son rythme.
# grgsm_decode -c relit record.cfile offline comme avant => SI preserve.
#
# Ouverture du FIFO en O_RDWR : ne bloque pas a l'ouverture ET ne voit jamais
# d'EOF (on garde nous-memes un write-end), donc qemu (O_WRONLY|O_NONBLOCK)
# trouve toujours un lecteur => son open reussit et le flux coule.
#
# =============================================================================
# [2026-08-11] CAPTURE COMPLETE, DECLENCHABLE — a cote du ring, pas a sa place
# =============================================================================
# Le ring ci-dessus rembobine tous les 128 Mo (~15,5 s) : parfait pour rejouer
# le passe immediat, inutilisable pour capturer un appel entier. La capture
# complete ci-dessous ne rembobine JAMAIS.
#
# POURQUOI UN TUBE DEDIE, et pas une lecture de iq_record.fifo : un FIFO n'est
# pas diffusable — deux lecteurs se PARTAGENT les octets, ils ne les voient pas
# chacun en entier. Draîner iq_record.fifo ici volerait donc une partie du flux
# au ring et casserait le decodage SI. On ajoute un 8e tube au relai
# (CALYPSO_RELAY_FIFOS), que qemu alimente comme les autres.
# ⚠️ RELAY_NFIFO_MAX vaut 8 dans qemu_wrap.c et 7 tubes etaient deja declares :
# ce tube consomme LA DERNIERE PLACE. En ajouter un 9e exige de relever la
# constante ET de recompiler, sinon il est ignore EN SILENCE.
#
# ⚠️ PAS SUR TMPFS. /dev/shm et /tmp sont de la RAM. Le flux fait 1083333 ech/s
# x 8 o = 8,7 Mo/s = 520 Mo/min = 31 Go/h : une capture complete sur tmpfs
# epuise la memoire de la machine en quelques minutes. Le defaut pointe donc sur
# l'overlay (disque). Verifier `df -h` avant une longue capture.
#
# DECLENCHEMENT : la capture n'ecrit que TANT QUE le fichier declencheur existe.
# touch /tmp/iq_full.trigger -> demarre (ouvre/tronque le .cfile)
# rm /tmp/iq_full.trigger -> arrete et FERME proprement le fichier
# Hors declenchement le tube est quand meme LU et jete : sinon l'anneau memoire
# du relai se remplit et qemu se met a jeter des trames entieres pour TOUS les
# tubes.
#
# PLAFOND : a CALYPSO_RECORD_FULL_MAX octets la capture S'ARRETE et le DIT. Elle
# ne rembobine pas — un fichier qui se remord la queue en silence est exactement
# ce qu'on veut eviter ici.
import os, sys, threading
FIFO = os.environ.get("CALYPSO_RECORD_FIFO", "/tmp/iq_record.fifo")
OUT = os.environ.get("CALYPSO_RECORD_FILE", "/tmp/record.cfile")
RING = int(os.environ.get("CALYPSO_RECORD_RING", str(128 << 20))) # 128 MB
FULL_ON = os.environ.get("CALYPSO_RECORD_FULL", "0") == "1"
FULL_FIFO = os.environ.get("CALYPSO_RECORD_FULL_FIFO", "/tmp/iq_full.fifo")
FULL_FILE = os.environ.get("CALYPSO_RECORD_FULL_FILE", "/opt/GSM/captures/full.cfile")
FULL_TRIGGER = os.environ.get("CALYPSO_RECORD_FULL_TRIGGER", "/tmp/iq_full.trigger")
FULL_MAX = int(os.environ.get("CALYPSO_RECORD_FULL_MAX", str(8 << 30))) # 8 GB
def log(msg):
sys.stderr.write(msg + "\n")
sys.stderr.flush()
def full_drain():
"""Capture complete : lit toujours, n'ecrit que sous declencheur, ne wrap jamais."""
if not os.path.exists(FULL_FIFO):
os.mkfifo(FULL_FIFO, 0o666)
os.makedirs(os.path.dirname(FULL_FILE) or ".", exist_ok=True)
fd = os.open(FULL_FIFO, os.O_RDWR) # meme idiome que le ring : ni blocage ni EOF
log("[record-full] %s -> %s (plafond %d MB) — declencheur : %s"
% (FULL_FIFO, FULL_FILE, FULL_MAX >> 20, FULL_TRIGGER))
out = None
w = 0
plafond_dit = False
while True:
b = os.read(fd, 1 << 16)
if not b:
continue
armed = os.path.exists(FULL_TRIGGER)
if armed and out is None:
out = open(FULL_FILE, "wb", buffering=1 << 20)
w = 0
plafond_dit = False
log("[record-full] DEMARRE -> %s" % FULL_FILE)
elif not armed and out is not None:
out.flush(); out.close(); out = None
log("[record-full] ARRETE : %s, %.1f MB (%.1f s d'IQ)"
% (FULL_FILE, w / 1048576.0, w / 8666664.0))
if out is None:
continue # on jette, mais on a LU : le relai ne bouche pas
if w + len(b) > FULL_MAX:
if not plafond_dit:
log("[record-full] PLAFOND %d MB atteint : capture ARRETEE (pas de "
"rembobinage). Retirer %s, vider %s, puis relancer."
% (FULL_MAX >> 20, FULL_TRIGGER, FULL_FILE))
plafond_dit = True
out.flush(); out.close(); out = None
continue
out.write(b)
w += len(b)
out.flush()
if not os.path.exists(FIFO):
os.mkfifo(FIFO, 0o666)
if FULL_ON:
threading.Thread(target=full_drain, daemon=True).start()
else:
log("[record-full] inactif (CALYPSO_RECORD_FULL != 1)")
fd = os.open(FIFO, os.O_RDWR) # O_RDWR : pas de blocage / pas d'EOF
out = open(OUT, "wb", buffering=1 << 20)
w = 0
log("[record-drain] %s -> %s (ring %d MB)" % (FIFO, OUT, RING >> 20))
while True:
b = os.read(fd, 1 << 16) # 64 KB
if not b:
continue
out.write(b)
w += len(b)
if w >= RING: # RING : rembobine (grgsm relit depuis 0)
out.seek(0); w = 0
out.flush()opt-gsm/si_bridge.py
opt-gsm/si_bridge.py
#!/usr/bin/env python3
# Bridge : grgsm_decode -v (DECODE le vrai SI du flux continu) -> parse les
# lignes SI (PD=0x06) -> GSMTAP 16B + L2 -> 4730 (shunt feed_si -> DATA_IND ->
# mobile). Le SCH reel (patch grgsm) -> UDP 4731 (feed_sb).
#
# GESTION DU FN (corrige le bug "GSMTAP frame_number=0") :
# - le FN reel vient des lignes SCHBSIC <bsic> <fn> <toa> (horloge fiable),
# ou d'un "FN: <n>" explicite si grgsm -v en imprime un.
# - on l'ecrit dans le champ frame_number du GSMTAP envoye a feed_si.
# - rappel 51-multiframe BCCH non-combine TS0 DL (FN % 51) :
# 0,10,20,30,40 = FCCH | 1,11,21,31,41 = SCH | 2-5 = BCCH
# 6-9,12-19,22-29,32-39,42-49 = CCCH (PCH/AGCH) | 50 = idle
# -> 31 = position SCH, 32 = debut d'un bloc CCCH, 51 = longueur multiframe.
import subprocess, socket, struct, re, sys, os, threading, time
ARFCN = int(os.environ.get("CALYPSO_CCCH_ARFCN", "514"))
CF = sys.argv[1] if len(sys.argv) > 1 else "/tmp/iq_grgsm.fifo"
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
SI = {0x1b:"SI3",0x1a:"SI2",0x1c:"SI4",0x19:"SI1",0x1d:"SI2bis",0x1e:"SI2ter",0x06:"RR"} # NB: 0x21 = PAGING (pas SI13=0x00)
# CCCH/AGCH downlink (PORTE 3a) : l'IMM ASSIGN doit etre forwarde au mobile, pas
# seulement les SI. gr-gsm le decode (06 3f) mais si_bridge ne forwardait que les SI.
CCCH = {0x3f:"IMM-ASSIGN", 0x39:"IMM-ASSIGN-EXT", 0x3a:"IMM-ASSIGN-REJ"}
# Paging Request (PCH) : type1=0x21 (== GSM48_MT_RR_PAG_REQ_1, PAS SI13), type2=0x22, type3=0x24.
PAGING = {0x21:"PAG-REQ-1", 0x22:"PAG-REQ-2", 0x24:"PAG-REQ-3"}
GSMTAP_BCCH = 0x01
GSMTAP_AGCH = 0x04
GSMTAP_SDCCH4 = 0x07
GSMTAP_SACCH = 0x07 | 0x80 # SDCCH/4 + ACCH = SACCH dediee SS0 (SI5/SI6 reels)
def fn51_role(fn):
m = fn % 51
if m in (0,10,20,30,40): return "FCCH"
if m in (1,11,21,31,41): return "SCH"
if m in (2,3,4,5): return "BCCH"
if m == 50: return "idle"
return "CCCH"
def gsmtap(fn, chan, l2):
# v2, hdr_len=4 (16B), type=UM(1), ts=0, arfcn, signal_dbm=0 (RIEN d'injecte),
# snr=0, frame_number=fn (REEL, decode), sub_type=chan, antenna=0, sub_slot=0, res=0
hdr = struct.pack(">BBBBHbbIBBBB", 2, 4, 0x01, 0, ARFCN & 0xffff,
0, 0, fn & 0xffffffff, chan, 0, 0, 0)
return hdr + l2
n = 0; nsch = 0; nsd = 0; nsa = 0; last_fn = 0; last_sd_fn = -1; last_sa_fn = -1
def handle_line(line):
global n, nsch, nsd, nsa, last_fn, last_sd_fn, last_sa_fn
# --- SCH reel : "SCHBSIC <bsic> <fn> <toa>" -> 4731 + horloge FN ---
msch = re.search(r"SCHBSIC\s+(\d+)\s+(\d+)\s+(-?\d+)", line)
if msch:
bsic, fn, toa = int(msch.group(1)), int(msch.group(2)), int(msch.group(3))
last_fn = fn
s.sendto(b"SCH1" + struct.pack("<iii", bsic, fn, toa), ("127.0.0.1", 4731))
nsch += 1
if nsch <= 10 or nsch % 100 == 0:
print("[si-bridge] SCH -> 4731 : BSIC=%d (ncc=%d bcc=%d) FN=%d (%%51=%d %s) TOA=%d #%d"
% (bsic,(bsic>>3)&7,bsic&7,fn,fn%51,fn51_role(fn),toa,nsch), flush=True)
return
# --- FN explicite eventuel dans la ligne -v ---
mfn = re.search(r"\bFN[:=]?\s*(\d+)", line)
if mfn: last_fn = int(mfn.group(1))
# --- ligne SI hexa ---
lstr = line.strip()
m = re.search(r":\s+([0-9a-fA-F][0-9a-fA-F ]+)$", lstr)
if not m: return
try: by = bytes(int(x,16) for x in m.group(1).split())
except: return
if len(by) >= 3 and by[1] == 0x06 and (by[2] in SI or by[2] in CCCH or by[2] in PAGING):
mline_fn = re.match(r"^\s*(\d+)\b", lstr)
si_fn = int(mline_fn.group(1)) if mline_fn else last_fn
L2 = (by[:23] + b"\x2b"*23)[:23]
# IMM ASSIGN (AGCH) ET Paging Request (PCH) -> meme chemin AGCH/PCH cote shunt
# (feed_agch -> bloc CCCH -> firmware chan_nr=0x90 -> gsm48_rr_rx_pch_agch qui
# dispatch IMM-ASS vs PAGING par msg type). SI -> BCCH.
is_ccch = (by[2] in CCCH) or (by[2] in PAGING)
chan = GSMTAP_AGCH if is_ccch else GSMTAP_BCCH
s.sendto(gsmtap(si_fn, chan, L2), ("127.0.0.1", 4730))
n += 1
name = CCCH.get(by[2]) or PAGING.get(by[2]) or SI.get(by[2])
if is_ccch or n <= 20 or n % 50 == 0:
print("[si-bridge] %s (mt=0x%02x) FN=%d (%%51=%d %s) -> %s (4730) #%d"
% (name, by[2], si_fn, si_fn%51, fn51_role(si_fn),
"feed_agch/PCH" if is_ccch else "feed_si", n), flush=True)
# --- SDCCH/4 SS0 DL (#2) : LAPDm (UA/AUTH), pas du RR -> gate fn%51 in {22-25} ---
mfn_sd = re.match(r"^\s*(\d+)\b", lstr)
sd_fn = int(mfn_sd.group(1)) if mfn_sd else last_fn
if len(by) >= 3 and (sd_fn % 51) in (22, 23, 24, 25):
L2 = (by[:23] + b"\x2b"*23)[:23]
# DEDUP PAR FN (corrige) : gr-gsm peut re-decoder le MEME bloc on-air (meme FN)
# -> il faut sauter ce doublon (re-decode). MAIS la BTS RE-EMET legitimement le
# MEME contenu (UA byte-identique) sur un FN DIFFERENT en reponse a une
# retransmission SABM T200 du mobile : ce re-UA DOIT etre forwarde sinon le
# mobile ne confirme jamais (UA rate -> boucle SABM -> "SABM not expected in
# timer recovery"). L'ancien dedup-par-contenu (L2 != last_sd_l2) DROPPAIT ce
# re-UA legitime. On dedup donc par FN decode (sd_fn) : grgsm emet chaque FN
# on-air UNE seule fois (FN monotone, verifie) -> dedup-par-FN supprime le
# re-decode (meme FN) ET forwarde le re-envoi BTS (FN distinct, meme contenu).
# On saute toujours le bourrage LAPDm (UI c=0x03).
ctrl = by[1] if len(by) > 1 else 0
is_fill = (ctrl == 0x03)
if (not is_fill) and sd_fn != last_sd_fn:
last_sd_fn = sd_fn
s.sendto(gsmtap(sd_fn, GSMTAP_SDCCH4, L2), ("127.0.0.1", 4730))
nsd += 1
if nsd <= 20 or nsd % 50 == 0:
print("[si-bridge] SDCCH/4 SS0 DL (a0=0x%02x c=0x%02x) FN=%d (%%51=%d) -> feed_sdcch (4730) #%d [new]"
% (by[0], ctrl, sd_fn, sd_fn%51, nsd), flush=True)
# ARMEMENT DU TCH : l'ASSIGNMENT COMMAND passe par ce meme bloc DL.
check_assignment(by, sd_fn)
# --- SACCH (SDCCH/4 SS0) DL : SI5(0x1d)/SI6(0x1e) REELS -> feed_sacch ---
# Slots SACCH SS0 (combine CCCH+SDCCH/4) : fn%51 in {42-45}. grgsm donne le
# bloc 23o = [L1 hdr 2][LAPDm: 03 03 len 06 mt L3...]. On detecte le RR header
# "06 1d"/"06 1e" et on forwarde le bloc TEL QUEL au shunt (sub_type 0x87 =
# SDCCH4|ACCH) -> feed_sacch REEL, remplace la fabrication SI3->SI6.
if len(by) >= 8 and (sd_fn % 51) in (42, 43, 44, 45):
mt_sa = None
for off in range(2, min(9, len(by) - 1)):
if by[off] == 0x06 and by[off + 1] in (0x1d, 0x1e):
mt_sa = by[off + 1]; break
if mt_sa is not None and sd_fn != last_sa_fn:
last_sa_fn = sd_fn
L2 = (by[:23] + b"\x2b" * 23)[:23]
s.sendto(gsmtap(sd_fn, GSMTAP_SACCH, L2), ("127.0.0.1", 4730))
nsa += 1
if nsa <= 20 or nsa % 50 == 0:
print("[si-bridge] SACCH SI%d (mt=0x%02x) FN=%d (%%51=%d) -> feed_sacch REEL (4730) #%d"
% (5 if mt_sa == 0x1d else 6, mt_sa, sd_fn, sd_fn % 51, nsa), flush=True)
# === STAGE 3 : DECHIFFREMENT DL — DEUX grgsm QUI SE DELEGUENT ================
# PROBLEME resolu : tuer grgsm pour le relancer avec -k au CIPHER MODE retirait
# le lecteur de la FIFO -> l'ipc-device se figeait (DL FIFO plein -> osmo-trx
# SETPOWER no-response -> feed_iq gele -> pas de LU accept).
# SOLUTION (design B) : on ne tue JAMAIS le grgsm CLAIR. Deux process tournent
# en parallele, chacun sur SA FIFO (les deux alimentees par l'ipc-device relay) :
# - grgsm CLAIR sur CF (iq_grgsm.fifo) : decode etablissement/ID/AUTH(RAND)/
# CIPHER MODE COMMAND. Ne sort QUE des trames CRC-valides -> se tait tout seul
# sur le dedie une fois le DL chiffre.
# - grgsm CHIFFRE sur CIPH_CF (-e -k) : spawne des qu'un Kc apparait (fenetre
# RAND->cipher : il a le temps de locker FCCH/SCH AVANT que le DL passe chiffre),
# decode le dedie chiffre (LU ACCEPT...). Tue seulement au release (Kc efface).
# Les deux sorties sont fusionnees par handle_line (sous lock) : le CRC de grgsm
# fait le tri, le dedup par FN evite les doublons. ZERO trou de lecteur sur la
# FIFO clair -> l'ipc-device ne gele plus. Le churn (spawn/kill) du grgsm chiffre
# est inoffensif grace au writer ipc-device non-bloquant (relay_fifo_writer).
CIPH_CF = os.environ.get("CALYPSO_CIPH_FIFO", "/tmp/iq_grgsm_ciph.fifo")
_hl_lock = threading.Lock()
# grgsm_decode HARDCODE le port GSMTAP UDP 4729 (socket_pdu UDP_SERVER+CLIENT, pas
# d'option). Deux instances -> "bind: Address already in use" -> le 2e crash
# (silencieux car stderr=DEVNULL) -> jamais de decode chiffre. On lit le STDOUT
# (-v), pas le 4729 -> on patche une COPIE pour le grgsm chiffre sur un port libre
# (4799). Genere au demarrage -> reproductible, survit au reclone du conteneur.
import shutil
def make_ciph_binary():
src = shutil.which("grgsm_decode")
if not src:
return "grgsm_decode"
dst = "/tmp/grgsm_decode_ciph"
try:
with open(src) as f:
code = f.read()
code = code.replace('"4729"', '"4799"') # port GSMTAP unique (pas de conflit avec le clair)
with open(dst, "w") as f:
f.write(code)
os.chmod(dst, 0o755)
print("[si-bridge] grgsm chiffre patche -> %s (GSMTAP 4729->4799)" % dst, flush=True)
return dst
except Exception as e:
print("[si-bridge] patch grgsm chiffre echoue (%s) -> fallback partage 4729" % e, flush=True)
return src
GRGSM_CIPH = make_ciph_binary()
def spawn_grgsm(fifo, algo=0, kc_hex=None, binary="grgsm_decode", errlog=None):
cmd = [binary, "-m", "BCCH_SDCCH4", "-t", "0", "-a", str(ARFCN),
"-c", fifo, "-s", "1083333", "-v"]
if kc_hex and algo in (1, 2, 3):
cmd += ["-e", str(algo), "-k", kc_hex]
err = open(errlog, "w") if errlog else subprocess.DEVNULL
return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=err, text=True)
# =============================================================================
# ARMEMENT DU TCH : ASSIGNMENT COMMAND -> calypso_tch_cfg -> 3e grgsm_decode
# =============================================================================
# [2026-08-11] RECONSTRUIT. Cette logique existait (08/08) et a disparu de
# l'arbre le 11/08. Sans elle la panne est en cascade et SILENCIEUSE :
# calypso_tch_cfg reste a 0 octet -> aucun `grgsm_decode -m TCHF` -> aucun
# GSMTAP dedie -> feed_tch_sacch JAMAIS appele -> la SACCH du TCH n'est pas
# alimentee -> le MS y lit le SI5/SI6 du camp (a_cd est PARTAGE camp<->TCH,
# avec DEUX conventions d'en-tete) -> "Short header message type 0x07
# unsupported" une fois par seconde, a partir de ~30 s apres l'ASSIGNMENT.
#
# CONTRAT DU FICHIER — lu chez son CONSOMMATEUR (qemu_wrap.c
# calypso_tch_cfg_read), jamais devine :
# 0..3 seq uint32 LE (≠0) 4 TN 5 TSC 6..7 ARFCN uint16 LE (16 o)
# Le seq est ecrit EN DERNIER : publication atomique, le lecteur ne voit jamais
# un demi-enregistrement.
#
# JUGES : calypso_tch_cfg non vide ; un process `-m TCHF` vivant ;
# `feed_tch_sacch` > 0 dans qemu.log ; `Short header 0x07` = 0.
TCH_CFG_PATH = "/dev/shm/calypso_tch_cfg"
TCH_FIFO = os.environ.get("CALYPSO_TCH_FIFO", "/tmp/iq_grgsm_tch.fifo")
TCH_SPEECH = "/dev/shm/calypso_tch_speech.gsm"
_tch_proc = None
_tch_cur = None # (tn, tsc, arfcn) actuellement arme
_tch_seq = 0
def _tch_write_cfg(tn, tsc, arfcn):
"""Publie TN/TSC/ARFCN. seq en dernier = publication atomique."""
global _tch_seq
_tch_seq += 1
buf = bytearray(16)
buf[4] = tn & 0xff
buf[5] = tsc & 0xff
buf[6:8] = int(arfcn).to_bytes(2, "little")
buf[0:4] = _tch_seq.to_bytes(4, "little")
fd = os.open(TCH_CFG_PATH, os.O_CREAT | os.O_WRONLY, 0o644)
try:
os.pwrite(fd, bytes(buf), 0)
finally:
os.close(fd)
def make_tch_binary():
"""Copie patchee de grgsm_decode pour le TCH — DEUX ports a corriger.
grgsm_decode code en dur 127.0.0.1:4729 DEUX fois (l.122-123) :
- l.122 UDP_SERVER : ne sert QU'A eviter les ICMP port-unreachable. Deux
instances -> « bind: Address already in use » -> la 2e MEURT. C'est ce
qui faisait mourir le decodeur TCH aussitot lance (14 reamorcages
mesures le 11/08), donc feed_tch_sacch = 0, donc SACCH dediee muette.
- l.123 UDP_CLIENT : c'est la VRAIE sortie GSMTAP. En mode TCHF,
tch_f_decoder ET cch_decoder y publient (l.213-214).
On bind donc le serveur sur un port libre, et on pointe le client droit sur
4730, que le shunt ecoute : il dispatche par sub_type — TCH_F (0x02) ->
feed_facch, TCH_ACCH (0x82) -> feed_tch_sacch (calypso_dsp_shunt.c:3126-3134).
Pas de parsing de stdout : les donnees vont du decodeur au shunt en direct.
⚠️ Remplacements CIBLES ligne par ligne : un `replace("4729","4730")` global
ferait binder le serveur sur 4730 et entrerait en conflit avec le shunt.
"""
src = shutil.which("grgsm_decode")
if not src:
return "grgsm_decode"
dst = "/tmp/grgsm_decode_tch"
try:
with open(src) as f:
code = f.read()
a = '"UDP_SERVER", "127.0.0.1", "4729"'
b = '"UDP_CLIENT", "127.0.0.1", "4729"'
if a not in code or b not in code:
print("[si-bridge] grgsm TCH : motifs de port introuvables -> "
"fallback non patche (conflit 4729 probable)", flush=True)
return src
code = code.replace(a, '"UDP_SERVER", "127.0.0.1", "4739"')
code = code.replace(b, '"UDP_CLIENT", "127.0.0.1", "4730"')
with open(dst, "w") as f:
f.write(code)
os.chmod(dst, 0o755)
print("[si-bridge] grgsm TCH patche -> %s (bind 4739, GSMTAP -> 4730)" % dst,
flush=True)
return dst
except Exception as e:
print("[si-bridge] patch grgsm TCH echoue (%s)" % e, flush=True)
return src
GRGSM_TCH = make_tch_binary()
def _tch_spawn(tn, arfcn):
"""3e decodeur, sur le tube IQ dedie (alimente par CALYPSO_RELAY_FIFOS)."""
cmd = [GRGSM_TCH, "-m", "TCHF", "-t", str(tn), "-a", str(arfcn),
"-c", TCH_FIFO, "-s", "1083333", "-v", "-d", "FR", "-o", TCH_SPEECH]
try:
err = open("/tmp/osmo-nitb/logs/grgsm_decode_tch.log", "w")
except Exception:
err = subprocess.DEVNULL
return subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=err, text=True)
def arm_tch(tn, tsc, arfcn):
global _tch_proc, _tch_cur
cfg = (tn, tsc, arfcn)
vivant = _tch_proc is not None and _tch_proc.poll() is None
if cfg == _tch_cur and vivant:
return # deja arme sur cette config
if _tch_proc is not None and vivant:
try: _tch_proc.terminate()
except Exception: pass
_tch_write_cfg(tn, tsc, arfcn)
_tch_proc = _tch_spawn(tn, arfcn)
_tch_cur = cfg
print("[si-bridge] TCH arme : TN=%d TSC=%d ARFCN=%d -> %s + grgsm -m TCHF (pid %s)"
% (tn, tsc, arfcn, TCH_CFG_PATH, _tch_proc.pid), flush=True)
def check_assignment(by, fn):
"""Detecte un ASSIGNMENT COMMAND (RR PD=0x06, MT=0x2e) sur le SDCCH DL.
Channel Description 2 (GSM 04.08 10.5.2.5), 3 octets, decodage aligne sur
`struct gsm48_chan_desc` de libosmocore :
b0 : chan_nr -> TN = b0 & 0x07
b1 : tsc(3) h(1) arfcn_high(2) spare(2)
TSC = (b1>>5)&7 H = (b1>>4)&1 ARFCN_high = b1 & 0x03
b2 : ARFCN_low -> ARFCN = (ARFCN_high<<8) | b2 [si H=0]
"""
for off in range(2, min(7, len(by) - 5)):
if by[off] != 0x06 or by[off + 1] != 0x2e:
continue
b0, b1, b2 = by[off + 2], by[off + 3], by[off + 4]
tn = b0 & 0x07
tsc = (b1 >> 5) & 0x07
h = (b1 >> 4) & 0x01
if h:
# ⚠️ REFUS EXPLICITE. Cette chaine ne gere PAS le saut de frequence
# (cf. la meme limite cote injection UL). Fabriquer une ARFCN a
# partir des champs MAIO/HSN donnerait un nombre FAUX en silence,
# et le decodeur ecouterait a cote sans que rien ne le signale.
print("[si-bridge] ASSIGNMENT COMMAND avec SAUT DE FREQUENCE (H=1) : "
"TCH NON arme (chaine sans hopping). FN=%d" % fn, flush=True)
return
arfcn = ((b1 & 0x03) << 8) | b2
print("[si-bridge] ASSIGNMENT COMMAND FN=%d -> chan_nr=0x%02x TN=%d TSC=%d ARFCN=%d"
% (fn, b0, tn, tsc, arfcn), flush=True)
arm_tch(tn, tsc, arfcn)
return
def reader_loop(proc, tag):
# log brut par grgsm (diagnostic decipher) : /root/grgsm_clair.raw / _ciph.raw
try: raw = open("/root/grgsm_%s.raw" % tag, "w")
except Exception: raw = None
for line in proc.stdout:
if raw:
try: raw.write(line); raw.flush()
except Exception: pass
with _hl_lock:
handle_line(line)
if raw:
try: raw.close()
except Exception: pass
print("[si-bridge] grgsm %s termine (n=%d nsch=%d nsd=%d nsa=%d)"
% (tag, n, nsch, nsd, nsa), flush=True)
def read_kc():
try:
with open("/dev/shm/calypso_kc", "rb") as f:
b = f.read(32)
except OSError:
return (0, 0, None)
if len(b) < 14:
return (0, 0, None)
seq = int.from_bytes(b[0:4], "little")
if seq == 0:
return (0, 0, None)
return (seq, b[4], b[6:14].hex())
# --- grgsm CLAIR : toujours vivant (watchdog respawn si crash), JAMAIS tue au cipher ---
clear = {"proc": None}
def ensure_clear():
p = clear["proc"]
if p is None or p.poll() is not None:
clear["proc"] = spawn_grgsm(CF)
threading.Thread(target=reader_loop, args=(clear["proc"], "clair"),
daemon=True).start()
print("[si-bridge] grgsm CLAIR spawn sur %s (jamais tue au cipher)" % CF, flush=True)
ensure_clear()
# --- grgsm CHIFFRE : spawne/tue selon le Kc, TOUJOURS sur CIPH_CF (jamais sur CF) ---
ciph = {"applied": (0, None), "proc": None}
while True:
time.sleep(0.5)
ensure_clear() # watchdog du grgsm clair
seq, algo, kc = read_kc()
want = (algo, kc) if (seq and algo in (1, 2, 3)) else (0, None)
if want == ciph["applied"]:
continue
# etat Kc change -> on (re)cycle UNIQUEMENT le grgsm chiffre (sur CIPH_CF)
p = ciph["proc"]
if p is not None:
try: p.kill(); p.wait(timeout=2)
except Exception: pass
ciph["proc"] = None
ciph["applied"] = want
if want[0]:
np = spawn_grgsm(CIPH_CF, algo=want[0], kc_hex=want[1], binary=GRGSM_CIPH,
errlog="/root/grgsm_ciph.err")
ciph["proc"] = np
threading.Thread(target=reader_loop, args=(np, "chiffre"),
daemon=True).start()
print("[si-bridge] grgsm CHIFFRE spawn sur %s : A5/%d kc=%s (decipher DL)"
% (CIPH_CF, want[0], want[1]), flush=True)
else:
print("[si-bridge] Kc efface -> grgsm chiffre arrete (clair seul)", flush=True)pont/airmesh.py
pont/airmesh.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# airmesh.py - le maillage se fait SUR LES BURSTS, plus sur les appels.
#
# ─────────────────────────────────────────────────────────────────────────────
# CE QUE CA CHANGE
#
# Jusqu'ici deux noeuds du banc se parlaient tout en haut de la pile : un trunk
# SIP entre deux Asterisk, un relais SMS en TCP, et du M3UA vers un hub. Chaque
# noeud avait sa propre radio, hermetique : le mobile de l'operateur 1 ne
# POUVAIT PAS entendre la BTS de l'operateur 2, parce que les bursts des deux
# ne se croisaient nulle part. Ce qu'on appelait "roaming" etait un
# raccordement telephonique entre deux reseaux qui s'ignoraient.
#
# Ici, ce qui traverse le reseau, c'est le burst. Le milieu radio devient
# commun : la BTS du noeud 2 emet, le mobile du noeud 1 l'entend, se cale
# dessus, lit ses SI, et fait sa mise a jour de localisation POUR DE VRAI. Tout
# le reste - BSC, MSC, HLR, MGW, Asterisk - reste enferme dans son noeud. Seul
# l'air est partage.
#
# ─────────────────────────────────────────────────────────────────────────────
# QUI FAIT QUOI : QEMU emet les bursts, fake_trx traduit, airmesh transporte
#
# Les bursts ne sont pas fabriques ici. Ils viennent du Calypso qui tourne dans
# QEMU : le firmware module, pont.py recupere ce que l'ARM depose dans ses
# sidebands, l'encode en bursts et le pose en TRXD. Dans l'autre sens il decode
# le descendant de la BTS et le rend au firmware. C'est LA la radio.
#
# fake_trx.py est le TRADUCTEUR de noeud : un commutateur de bursts qui relie N
# transceivers et distribue ce que l'un EMET a tous ceux dont la frequence de
# RECEPTION correspond (burst_fwd.py, forward_msg). Il ne fabrique rien, il
# aiguille.
#
# airmesh est le transport entre noeuds. Il se branche sur le traducteur, prend
# ce qui y passe, l'envoie aux pairs, et injecte ce qui en revient.
#
# ┌──────────────────────── un noeud ─────────────────────────┐
# │ QEMU/Calypso ──sidebands── pont.py ──TRXD 5700──┐ │
# │ osmo-bts-trx ───────────────────────TRXD 5700───┤ │
# │ trxcon/mobile ──────────────────────TRXD 6700───┤ fake │
# │ airmesh(cote mobile) ───────────────TRXD 6706───┤ _trx │
# │ airmesh(cote BTS) ──────────────────TRXD 6709───┘ │
# └───────────────────────────┬───────────────────────────────┘
# │ UDP AIRM (defaut 4739)
# ┌───────────────┴───────────────┐
# noeud 2 noeud 3
#
# POURQUOI DEUX BRANCHEMENTS. fake_trx apparie sur la FREQUENCE : un
# transceiver n'entend que ce qui est emis sur sa frequence de reception. Un
# seul point d'accroche n'entendrait donc qu'un sens, et un mobile qui n'entend
# pas la BTS ne s'y attache jamais.
# - cote mobile : rx = descendant, tx = montant. Il ENTEND ce que la BTS
# locale emet, et INJECTE le montant des mobiles distants vers elle.
# - cote BTS : rx = montant, tx = descendant. Il ENTEND ce que les mobiles
# locaux emettent, et INJECTE le descendant des BTS distantes vers eux.
#
# ─────────────────────────────────────────────────────────────────────────────
# COMMENT ON S'Y BRANCHE SANS MODIFIER UNE LIGNE DE QEMU-SRC
#
# fake_trx cree un transceiver de plus par mobile declare (60-fake-trx.sh :
# --trx ue<n>@127.0.0.<g>:<port>, port = 6700 + (n-1) * 3). Ces transceivers ne
# sont "des mobiles" que par leur nom : fake_trx ne leur connait ni role ni
# bande tant qu'ils n'ont pas envoye RXTUNE et TXTUNE.
#
# On declare donc DEUX mobiles de plus que le banc n'en a - N_MS + 2 - et
# airmesh prend leurs deux places. Il accorde la premiere comme un mobile et la
# seconde a l'envers, comme une BTS. Aucune modification du lanceur, aucun
# fichier de qosmo-grgsm touche : deux emplacements deja prevus, occupes autrement.
#
# CE QU'IL FAUT SAVOIR EN LES LISANT : `pgrep -a fake_trx` montrera deux
# mobiles qui n'existent pas, et le journal de fake_trx annoncera ue3 et ue4.
# C'est voulu. Le nom vient du lanceur, pas de nous.
#
# ─────────────────────────────────────────────────────────────────────────────
# LES TROIS DIFFICULTES, ET CE QUI EST FAIT DE CHACUNE
#
# 1. L'HORLOGE. Une trame TDMA dure 4,615 ms et porte un numero (FN) sur 22
# bits qui reboucle toutes les 3 h 28. Deux noeuds ont deux horloges sans
# rapport : un burst etiquete fn=12345 chez l'un ne designe aucun instant
# chez l'autre.
# CE QUI EST FAIT : un suivi d'ECART. airmesh mesure en continu la difference
# entre le FN local et le FN annonce par chaque pair, et REECRIT le FN a
# l'injection. Les deux horloges avancent au meme rythme nominal (un timer
# logiciel de 4,615 ms des deux cotes), l'ecart est donc quasi constant et
# la derive se rattrape a chaque burst recu.
# CE QUI SERAIT MIEUX : un noeud maitre qui DIFFUSE le FN, les autres
# n'ayant plus d'horloge propre. C'est la bonne reponse, elle demande de
# prendre la main sur clck_gen ; l'ecart suffit tant que les noeuds sont sur
# le meme LAN. La fonction _fn_offset() est le seul endroit a remplacer.
#
# 2. LE DELAI. Un burst doit arriver AVANT que l'ordonnanceur local n'en ait
# besoin. On l'injecte donc en avance de LOOKAHEAD trames (defaut 6, soit
# ~28 ms) : un aller-retour LAN tient largement dedans, une liaison a 30 ms
# de latence non. Les bursts arrives trop tard sont COMPTES, jamais jetes en
# silence - un maillage radio qui perd 5 % de ses bursts se voit dans les
# statistiques, pas dans les journaux du BSC.
#
# 3. QUI ENTEND QUI. Sans graphe, chaque mobile entendrait toutes les BTS du
# banc a pleine puissance : plus de cellule, plus de reselection, plus rien
# a demontrer. data/air-mesh.txt donne l'attenuation entre noeuds, en dB.
# Au-dela d'un plancher, le burst n'est pas transmis - c'est ce qui fait un
# MAILLAGE et non un bus.
#
# ─────────────────────────────────────────────────────────────────────────────
# LES DEUX MODES (AIRMESH_MODE)
#
# fake (defaut) - le burst traverse tel quel. C'est du commutation pure :
# aucune demodulation, aucun codage canal, le train de bits
# ressort identique. Exact et peu couteux.
# pont - le burst passe par la chaine du Calypso : demodulation en
# L2 (libosmocoding), remise a QEMU par les sidebands, et ce
# que l'ARM repond est re-module en bursts avant de repartir
# sur le maillage. C'est pont.py qui tient cette chaine ;
# airmesh lui passe la main par une FIFO et n'y touche pas.
#
# ─────────────────────────────────────────────────────────────────────────────
# CE QUI N'EST PAS VALIDE
#
# Ce fichier est ecrit contre les formats et les conventions de port lus dans
# osmocom-bb (trx_toolkit : transceiver.py, data_msg.py, burst_fwd.py) et
# contre le TRXDv0 deja implemente des deux cotes par pont.py. Il n'a PAS
# encore tourne sur le banc : la partie a eprouver en premier est le suivi
# d'ecart d'horloge (point 1), qui est la seule qui repose sur une hypothese -
# que les deux clck_gen avancent au meme rythme. Les compteurs sont la pour
# ca : "hors fenetre" qui monte, c'est le point 1 ou le point 2 qui lache.
import argparse, os, socket, struct, sys, threading, time, collections
# ── TRXD v0, tel qu'osmocom-bb l'ecrit ───────────────────────────────────────
# En-tete commun : octet 0 = (version << 4) | numero de slot, puis FN sur 4
# octets gros-boutiste. (data_msg.py, parse_hdr / append_hdr_to)
# TxMsg (L1 -> transceiver) : en-tete commun + pwr(1) = 6 octets
# RxMsg (transceiver -> L1) : en-tete commun + rssi(1) + toa(2) = 8 octets
# Puis 148 octets, un par bit (durs a l'emission, souples a la reception).
TRXD_VER = 0
CHDR_LEN = 5
TX_HDR_LEN = CHDR_LEN + 1
RX_HDR_LEN = CHDR_LEN + 3
BURST_LEN = 148
# ── Notre protocole entre noeuds ─────────────────────────────────────────────
# Volontairement plat et lisible au tcpdump : on debogue un maillage radio en
# regardant passer les paquets, pas en relisant un schema de serialisation.
# magie(4) ver(1) sens(1) noeud(2) arfcn(2) tn(1) reserve(1)
# fn(4) pwr(1) rssi(1) toa(2 signe) nbits(2) = 22 octets, puis nbits octets
AIRM_MAGIC = b"AIRM"
AIRM_VER = 1
AIRM_HDR = ">4sBBHHBBIbbhH"
AIRM_HDR_LEN = struct.calcsize(AIRM_HDR)
DIR_DL = 0 # ce qu'une BTS emet
DIR_UL = 1 # ce qu'un mobile emet
FN_MAX = 2715648 # 26 * 51 * 2048, le rebouclage du FN GSM
FRAME_S = 4.615e-3
def arfcn_to_khz(arfcn, uplink=False):
"""Frequence en kHz. GSM900 / DCS1800, les deux bandes du banc.
fake_trx n'apparie pas sur l'ARFCN mais sur la FREQUENCE : se tromper de
bande ne donne pas une erreur, cela donne un transceiver qui n'entend
simplement jamais rien - la panne la plus longue a trouver.
"""
if 512 <= arfcn <= 885: # DCS 1800
dl = 1805200 + 200 * (arfcn - 512)
return dl - 95000 if uplink else dl
if 0 <= arfcn <= 124: # GSM 900
dl = 935000 + 200 * arfcn
return dl - 45000 if uplink else dl
if 975 <= arfcn <= 1023: # E-GSM 900
dl = 935000 + 200 * (arfcn - 1024)
return dl - 45000 if uplink else dl
raise ValueError("ARFCN hors des bandes connues : %d" % arfcn)
class Stats:
"""Ce qu'on regarde quand le maillage "marche a moitie"."""
def __init__(self):
self.lock = threading.Lock()
self.c = collections.Counter()
def bump(self, k, n=1):
with self.lock:
self.c[k] += n
def render(self):
with self.lock:
return " ".join("%s=%d" % (k, v) for k, v in sorted(self.c.items()))
ST = Stats()
class Topologie:
"""Qui entend qui, et a quel prix.
data/air-mesh.txt, une arete par ligne :
<noeud a> <noeud b> <attenuation dB> [uni]
Sans le mot "uni" l'arete vaut dans les deux sens. Une paire absente du
fichier ne s'entend pas : c'est un maillage, pas un bus - deux noeuds ne se
voient que si quelqu'un l'a ecrit.
"""
def __init__(self, chemin, defaut_db):
self.aretes = {}
self.defaut = defaut_db
self.ouvert = False
if chemin and os.path.isfile(chemin):
self._charger(chemin)
self.ouvert = True
def _charger(self, chemin):
with open(chemin, "r", encoding="utf-8", errors="replace") as f:
for ligne in f:
ligne = ligne.split("#", 1)[0].strip()
if not ligne:
continue
ch = ligne.split()
if len(ch) < 3:
continue
try:
a, b, att = int(ch[0]), int(ch[1]), float(ch[2])
except ValueError:
continue
self.aretes[(a, b)] = att
if len(ch) < 4 or ch[3] != "uni":
self.aretes[(b, a)] = att
def attenuation(self, src, dst):
"""dB entre deux noeuds, ou None si l'un n'entend pas l'autre.
Sans fichier, tout le monde s'entend a l'attenuation par defaut : c'est
le comportement d'un bus, utile pour un premier essai a deux noeuds, et
il est ANNONCE au demarrage pour qu'on ne le prenne pas pour une
topologie.
"""
if not self.ouvert:
return self.defaut
return self.aretes.get((src, dst))
class TrxLien:
"""Une L1 vue de fake_trx : CTRL et DATA, exactement comme trxcon.
Conventions de port lues dans transceiver.py (Transceiver.__init__) :
le transceiver BINDE base+1 (CTRL) et base+2 (DATA) ; la L1 binde base+101
et base+102 et emet vers les premiers. On est du cote L1.
"""
def __init__(self, nom, host, base, arfcn, rx_uplink):
self.nom = nom
self.host = host
self.base = base
self.arfcn = arfcn
self.rx_uplink = rx_uplink # True : on ecoute le montant
self.ctrl = self._sock(base + 101)
self.data = self._sock(base + 102)
self.ctrl_seq = 0
self.slots = set()
@staticmethod
def _sock(port):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("127.0.0.1", port))
return s
# ── CTRL ────────────────────────────────────────────────────────────────
def cmd(self, texte, attente=1.0):
"""Envoie une commande CTRL et attend sa reponse.
On ATTEND la reponse plutot que de tirer les commandes a la suite :
fake_trx repond RSP <cmd> <code>, et un code non nul (frequence hors
bande, slot refuse) passait inapercu. Un transceiver mal accorde
n'emet aucune erreur ensuite - il se tait, simplement.
"""
self.ctrl.sendto(("CMD " + texte + "\0").encode(), (self.host, self.base + 1))
self.ctrl.settimeout(attente)
try:
rep, _ = self.ctrl.recvfrom(512)
except socket.timeout:
return None
finally:
self.ctrl.settimeout(None)
return rep.rstrip(b"\0").decode(errors="replace")
def accorder(self, slots):
"""POWERON complet : accord des deux frequences puis ouverture des slots.
L'ordre compte. fake_trx refuse SETSLOT sur un transceiver qui n'a pas
de frequence, et POWERON sur un transceiver sans slot ne sert a rien :
il tourne et n'entend rien.
"""
rx = arfcn_to_khz(self.arfcn, uplink=self.rx_uplink)
tx = arfcn_to_khz(self.arfcn, uplink=not self.rx_uplink)
for cmd in ("RXTUNE %d" % rx, "TXTUNE %d" % tx):
rep = self.cmd(cmd)
if rep is None or " 0" not in rep:
raise RuntimeError("%s : %s refuse (%s)" % (self.nom, cmd, rep))
for tn in slots:
# type 1 = TCH/F pour fake_trx ; ce qui compte ici est que le slot
# soit ACTIF, le contenu reel est porte par les bursts eux-memes.
self.cmd("SETSLOT %d 1" % tn)
self.slots.add(tn)
rep = self.cmd("POWERON")
if rep is None or " 0" not in rep:
raise RuntimeError("%s : POWERON refuse (%s)" % (self.nom, rep))
def eteindre(self):
try:
self.cmd("POWEROFF", attente=0.3)
except OSError:
pass
# ── DATA ────────────────────────────────────────────────────────────────
def lire(self):
"""Un burst que fake_trx nous livre : ce que le voisin local a EMIS."""
paquet, _ = self.data.recvfrom(1024)
if len(paquet) < RX_HDR_LEN + BURST_LEN:
ST.bump("rx_court")
return None
ver = paquet[0] >> 4
if ver != TRXD_VER:
ST.bump("rx_version_inattendue")
return None
tn = paquet[0] & 0x0F
fn = struct.unpack(">L", paquet[1:5])[0]
rssi = -paquet[5]
toa = struct.unpack(">h", paquet[6:8])[0]
bits = paquet[RX_HDR_LEN:RX_HDR_LEN + BURST_LEN]
return tn, fn, rssi, toa, bits
def ecrire(self, tn, fn, pwr, bits):
"""Injecte un burst : fake_trx le distribuera a qui est accorde dessus."""
hdr = bytearray(TX_HDR_LEN)
hdr[0] = (TRXD_VER << 4) | (tn & 0x0F)
hdr[1:5] = struct.pack(">L", fn % FN_MAX)
hdr[5] = pwr & 0xFF
self.data.sendto(bytes(hdr) + bytes(bits), (self.host, self.base + 2))
class Maillage:
def __init__(self, cfg):
self.cfg = cfg
self.noeud = cfg.node
self.topo = Topologie(cfg.topologie, cfg.attenuation_defaut)
self.pairs = cfg.pairs # [(id, ip, port)]
self.stop = threading.Event()
# L'horloge : FN local vu en dernier, et ecart estime par pair.
self._fn_local = 0
self._fn_vu_a = time.monotonic()
self._ecart = {} # id pair -> ecart en trames
self._verrou = threading.Lock()
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind((cfg.bind, cfg.port))
# Cote mobile : entend le descendant local, injecte le montant distant.
self.ms = TrxLien("airmesh-ms", "127.0.0.1", cfg.trx_ms,
cfg.arfcn, rx_uplink=False)
# Cote BTS : entend le montant local, injecte le descendant distant.
self.bts = TrxLien("airmesh-bts", "127.0.0.1", cfg.trx_bts,
cfg.arfcn, rx_uplink=True)
# ── Horloge ─────────────────────────────────────────────────────────────
def _note_fn_local(self, fn):
with self._verrou:
self._fn_local = fn
self._fn_vu_a = time.monotonic()
def _fn_maintenant(self):
"""Le FN local, extrapole depuis le dernier burst vu.
On n'a pas d'horloge a nous : le seul FN de reference est celui que
fake_trx nous livre avec chaque burst. Entre deux bursts on avance a la
montre - 4,615 ms par trame. C'est suffisant pour une avance de six
trames ; ce serait insuffisant pour dater quoi que ce soit.
"""
with self._verrou:
base, quand = self._fn_local, self._fn_vu_a
return (base + int((time.monotonic() - quand) / FRAME_S)) % FN_MAX
def _fn_offset(self, pair, fn_distant):
"""Ecart entre l'horloge d'un pair et la notre, lisse.
C'est LE point fragile du maillage, et le seul endroit a remplacer le
jour ou un noeud maitre diffusera le FN : ici, on suppose seulement que
les deux horloges avancent au meme rythme.
"""
vu = (self._fn_maintenant() - fn_distant) % FN_MAX
with self._verrou:
ancien = self._ecart.get(pair)
if ancien is None:
self._ecart[pair] = vu
return vu
# Moyenne glissante sur l'ecart circulaire : sans le repliement,
# un rebouclage du FN ferait sauter l'ecart de deux millions de
# trames et le maillage se tairait pendant trois heures.
d = (vu - ancien + FN_MAX // 2) % FN_MAX - FN_MAX // 2
neuf = (ancien + d // 8) % FN_MAX
self._ecart[pair] = neuf
return neuf
# ── Emission vers les pairs ─────────────────────────────────────────────
def _diffuser(self, sens, tn, fn, rssi, toa, bits):
for (pid, ip, port) in self.pairs:
att = self.topo.attenuation(self.noeud, pid)
if att is None:
ST.bump("hors_portee")
continue
r = max(-120, min(-1, int(rssi - att)))
if r <= self.cfg.plancher:
ST.bump("sous_plancher")
continue
paquet = struct.pack(AIRM_HDR, AIRM_MAGIC, AIRM_VER, sens,
self.noeud, self.cfg.arfcn, tn, 0,
fn % FN_MAX, 0, r, toa, len(bits)) + bytes(bits)
try:
self.sock.sendto(paquet, (ip, port))
ST.bump("emis_dl" if sens == DIR_DL else "emis_ul")
except OSError:
ST.bump("emission_echec")
def _boucle_locale(self, lien, sens):
"""Ce que le voisin LOCAL emet part vers les pairs."""
while not self.stop.is_set():
try:
lu = lien.lire()
except OSError:
break
if lu is None:
continue
tn, fn, rssi, toa, bits = lu
self._note_fn_local(fn)
self._diffuser(sens, tn, fn, rssi, toa, bits)
# ── Reception depuis les pairs ──────────────────────────────────────────
def _boucle_maillage(self):
while not self.stop.is_set():
try:
paquet, _ = self.sock.recvfrom(2048)
except OSError:
break
if len(paquet) < AIRM_HDR_LEN:
ST.bump("mesh_court")
continue
(magie, ver, sens, src, arfcn, tn, _res,
fn, _pwr, rssi, toa, nbits) = struct.unpack(AIRM_HDR, paquet[:AIRM_HDR_LEN])
if magie != AIRM_MAGIC or ver != AIRM_VER:
ST.bump("mesh_etranger")
continue
bits = paquet[AIRM_HDR_LEN:AIRM_HDR_LEN + nbits]
if len(bits) != nbits or nbits != BURST_LEN:
ST.bump("mesh_tronque")
continue
if arfcn != self.cfg.arfcn:
# Un pair sur une autre porteuse : c'est normal et c'est meme
# le but - deux operateurs voisins n'occupent pas le meme
# canal. On ne le compte pas comme une erreur.
ST.bump("autre_arfcn")
continue
# Ou ce burst tombe-t-il dans NOTRE temps.
fn_local = (fn + self._fn_offset(src, fn)) % FN_MAX
cible = (self._fn_maintenant() + self.cfg.lookahead) % FN_MAX
retard = (cible - fn_local + FN_MAX // 2) % FN_MAX - FN_MAX // 2
if retard > self.cfg.tolerance:
ST.bump("hors_fenetre_tard")
continue
fn_inj = cible
if sens == DIR_DL:
# Un descendant distant : c'est une BTS d'ailleurs. On l'injecte
# par le cote BTS, celui dont les mobiles locaux ecoutent le tx.
self.bts.ecrire(tn, fn_inj, self._pwr(rssi), bits)
ST.bump("injecte_dl")
else:
self.ms.ecrire(tn, fn_inj, self._pwr(rssi), bits)
ST.bump("injecte_ul")
@staticmethod
def _pwr(rssi):
"""TxMsg porte une attenuation (0 = pleine puissance), pas un RSSI.
On reporte donc l'attenuation deja appliquee par la topologie, bornee
a l'octet : c'est ce qui donne au mobile une raison de preferer une
cellule a une autre.
"""
return max(0, min(255, -rssi - 40))
# ── Vie du processus ────────────────────────────────────────────────────
def _boucle_stats(self):
while not self.stop.wait(self.cfg.stats_s):
ecarts = " ".join("n%d:%+d" % (k, v if v < FN_MAX // 2 else v - FN_MAX)
for k, v in sorted(self._ecart.items()))
print("[airmesh] fn=%d %s | ecarts %s" %
(self._fn_maintenant(), ST.render(), ecarts or "aucun"),
flush=True)
def demarrer(self):
print("[airmesh] noeud %d arfcn %d pairs %s" %
(self.noeud, self.cfg.arfcn,
", ".join("%d@%s:%d" % p for p in self.pairs) or "aucun"), flush=True)
if not self.topo.ouvert:
print("[airmesh] AUCUNE topologie (%s) : tous les pairs s'entendent "
"a %.0f dB - c'est un bus, pas un maillage."
% (self.cfg.topologie, self.cfg.attenuation_defaut), flush=True)
if self.cfg.mode != "fake":
print("[airmesh] mode '%s' : les bursts passent par la chaine "
"Calypso de pont.py (FIFO %s)" % (self.cfg.mode, self.cfg.fifo),
flush=True)
slots = [int(x) for x in self.cfg.slots.split(",") if x != ""]
self.ms.accorder(slots)
self.bts.accorder(slots)
fils = [
threading.Thread(target=self._boucle_locale, args=(self.ms, DIR_DL),
name="local-dl", daemon=True),
threading.Thread(target=self._boucle_locale, args=(self.bts, DIR_UL),
name="local-ul", daemon=True),
threading.Thread(target=self._boucle_maillage, name="maillage", daemon=True),
threading.Thread(target=self._boucle_stats, name="stats", daemon=True),
]
for f in fils:
f.start()
try:
while not self.stop.is_set():
time.sleep(0.5)
except KeyboardInterrupt:
pass
finally:
self.stop.set()
self.ms.eteindre()
self.bts.eteindre()
print("[airmesh] arret. %s" % ST.render(), flush=True)
def paire(txt):
"""--pair 2:192.168.1.12:4739"""
ch = txt.split(":")
if len(ch) != 3:
raise argparse.ArgumentTypeError("attendu <noeud>:<ip>:<port>, recu %r" % txt)
return (int(ch[0]), ch[1], int(ch[2]))
def main():
p = argparse.ArgumentParser(
prog="airmesh",
description="Maillage de bursts GSM entre noeuds : le milieu radio devient commun.")
p.add_argument("--node", type=int, required=True, help="numero de CE noeud")
p.add_argument("--arfcn", type=int, default=int(os.environ.get("AIRMESH_ARFCN", "514")))
p.add_argument("--pair", dest="pairs", action="append", type=paire, default=[],
metavar="NOEUD:IP:PORT", help="un pair du maillage (repetable)")
p.add_argument("--bind", default=os.environ.get("AIRMESH_BIND", "0.0.0.0"))
p.add_argument("--port", type=int, default=int(os.environ.get("AIRMESH_PORT", "4739")))
# 6700 + (n-1)*3 : les deux emplacements de mobile qui suivent les vrais.
# Avec deux MS reels, le banc doit tourner en N_MS=4 et airmesh prend 3 et 4.
p.add_argument("--trx-ms", type=int, default=int(os.environ.get("AIRMESH_TRX_MS", "6706")),
help="base TRXD de l'emplacement pris cote mobile (defaut : MS 3)")
p.add_argument("--trx-bts", type=int, default=int(os.environ.get("AIRMESH_TRX_BTS", "6709")),
help="base TRXD de l'emplacement pris cote BTS (defaut : MS 4)")
p.add_argument("--slots", default=os.environ.get("AIRMESH_SLOTS", "0,1,2,3,4,5,6,7"))
p.add_argument("--lookahead", type=int, default=int(os.environ.get("AIRMESH_LOOKAHEAD", "6")),
help="trames d'avance a l'injection (6 = ~28 ms)")
p.add_argument("--tolerance", type=int, default=int(os.environ.get("AIRMESH_TOL", "2")),
help="trames de retard tolerees avant de compter le burst perdu")
p.add_argument("--topologie", default=os.environ.get("AIRMESH_TOPO", "data/air-mesh.txt"))
p.add_argument("--attenuation-defaut", type=float,
default=float(os.environ.get("AIRMESH_ATT", "20")))
p.add_argument("--plancher", type=int, default=int(os.environ.get("AIRMESH_PLANCHER", "-110")),
help="dBm sous lesquels le burst n'est pas transmis")
# fake : le burst traverse tel quel, airmesh n'est qu'un transport.
# pont : il passe par la chaine Calypso de pont.py - demodulation, remise a
# QEMU, remodulation - avant de repartir. C'est le mode qui fait vraiment
# emettre le firmware pour le compte du maillage.
p.add_argument("--mode", choices=("fake", "pont"),
default=os.environ.get("AIRMESH_MODE", "fake"))
p.add_argument("--fifo", default=os.environ.get("AIRMESH_FIFO", "/dev/shm/airmesh_pont"))
p.add_argument("--stats-s", type=float, default=float(os.environ.get("AIRMESH_STATS", "10")))
cfg = p.parse_args()
try:
Maillage(cfg).demarrer()
except RuntimeError as e:
print("[airmesh] %s" % e, file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())pont/cipher.py
pont/cipher.py
import logging
import os
import struct
import threading
import time
from . import gsm
log = logging.getLogger("pont")
KC_PATH = "/dev/shm/calypso_kc_l1"
KC_TTL = 0.1
class Cipher:
def __init__(self, retention, stats):
self.retention = retention
self.stats = stats
self._lock = threading.Lock()
self._fd = None
self._cache = None
self._next = 0.0
self._held = None
# La BTS ne chiffre le descendant qu'apres avoir recu la premiere trame
# montante chiffree (osmo-bts l1sap.c check_for_first_ciphrd). Tant que
# ce n'est pas constate, le descendant du canal dedie est en clair.
self.dl_active = False
def _read(self):
now = time.monotonic()
if now < self._next:
return self._cache
self._next = now + KC_TTL
try:
if self._fd is None:
self._fd = os.open(KC_PATH, os.O_RDONLY)
b = os.pread(self._fd, 32, 0)
except OSError:
if self._fd is not None:
os.close(self._fd)
self._fd = None
self._cache = None
return None
if len(b) < 14:
self._cache = None
return None
algo = b[4]
kc = b[6:14]
self._cache = (algo, kc) if 1 <= algo <= 3 and any(kc) else None
return self._cache
def current(self):
with self._lock:
key = self._read()
if key:
self._held = key
return key
if self.retention and self._held:
return self._held
return None
def release(self, reason):
with self._lock:
if self._held is None:
return
self._held = None
self.dl_active = False
log.info("Kc lache (%s) : le lien repart en clair", reason)
def confirm_dl(self, fn):
with self._lock:
if self.dl_active:
return
self.dl_active = True
log.info("chiffrement descendant confirme par la BTS (fn=%u)", fn)
def apply(self, burst, fn, uplink):
key = self.current()
if key is None:
return burst
dl, ul = gsm.a5_keystream(key[0], key[1], fn)
if uplink:
self.stats.a5_ul += 1
else:
self.stats.a5_dl += 1
return gsm.a5_xor(burst, ul if uplink else dl)pont/config.py
pont/config.py
import argparse
import os
from dataclasses import dataclass
def _env(name, default):
return os.environ.get(name, default)
def _flag(name, default):
return _env(name, default) not in ("0", "", "no")
@dataclass(frozen=True)
class Config:
trx_bind: str
trx_base: int
arfcn: int
bsic: int
ul_fn_advance: int
window_tol: int
rssi: int
gsmtap_port: int
sch_port: int
tap: bool
tap_port: int
bsc_cfg: str
clock_period: int
kc_retention: bool
record: bool
record_ul: bool
record_dl_path: str
record_ul_path: str
record_dl_fifo: str
record_ul_fifo: str
record_max_mb: int
record_min_free_mb: int
record_osr: int
def parse(argv=None):
p = argparse.ArgumentParser(prog="pont", description="Pont TRX entre le BTS et la couche 1 gr-gsm de QEMU")
p.add_argument("--arfcn", type=int, default=int(_env("PONT_ARFCN", "514")))
p.add_argument("--bsic", type=int, default=int(_env("PONT_BSIC", "7")))
p.add_argument("--trx-bind", default=_env("PONT_TRX_BIND", "127.0.0.1"))
p.add_argument("--trx-base", type=int, default=int(_env("PONT_TRX_BASE", "5700")))
p.add_argument("--bsc-cfg", default=_env("PONT_BSC_CFG", "/etc/osmocom/osmo-bsc.cfg"))
p.add_argument("--no-tap", action="store_true", default=not _flag("PONT_TAP", "1"))
p.add_argument("--no-record", action="store_true", default=not _flag("PONT_AIRREC", "1"))
a = p.parse_args(argv)
return Config(
trx_bind=a.trx_bind,
trx_base=a.trx_base,
arfcn=a.arfcn,
bsic=a.bsic,
ul_fn_advance=int(_env("PONT_UL_FN_ADVANCE", "3")),
window_tol=int(_env("PONT_WINDOW_TOL", "1")),
rssi=int(_env("PONT_RSSI", "60")),
gsmtap_port=4730,
sch_port=4731,
tap=not a.no_tap,
tap_port=int(_env("PONT_TAP_PORT", "4729")),
bsc_cfg=a.bsc_cfg,
clock_period=int(_env("PONT_CLCK_PERIOD_FN", "51")),
kc_retention=_flag("PONT_KC_RETENTION", "1"),
record=not a.no_record,
record_ul=_flag("PONT_AIRREC_UL", "1"),
record_dl_path=_env("PONT_AIRREC_DL_PATH", "/root/record.cfile"),
record_ul_path=_env("PONT_AIRREC_UL_PATH", "/root/record_ul.cfile"),
record_dl_fifo=_env("PONT_AIRREC_DL_FIFO", "/tmp/iq_fft.fifo"),
record_ul_fifo=_env("PONT_AIRREC_UL_FIFO", "/tmp/iq_fft_ms.fifo"),
record_max_mb=int(_env("PONT_AIRREC_MAX_MB", "4096")),
record_min_free_mb=int(_env("PONT_AIRREC_MINFREE_MB", "8192")),
record_osr=int(_env("PONT_AIRREC_OSR", "4")),
)pont/downlink.py
pont/downlink.py
import logging
import os
import socket
import struct
from . import gsm
log = logging.getLogger("pont")
TCH_DL_PATH = "/dev/shm/calypso_tch_dl"
TCH_DL_SLOTS = 16
TCH_DL_SLOT = 48
GSMTAP_HOST = "127.0.0.1"
class Feeder:
def __init__(self, cfg, stats):
self.cfg = cfg
self.stats = stats
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
def l2(self, fn, chan, l2, tn=0, uplink=False):
pkt = gsm.gsmtap_header(self.cfg.arfcn, fn, chan, tn, uplink) + bytes(l2)
if not uplink:
self.sock.sendto(pkt, (GSMTAP_HOST, self.cfg.gsmtap_port))
if self.cfg.tap:
self.sock.sendto(pkt, (GSMTAP_HOST, self.cfg.tap_port))
def sch(self, fn):
self.sock.sendto(b"SCH2" + struct.pack("<iii", self.cfg.bsic, fn, self.cfg.arfcn), (GSMTAP_HOST, self.cfg.sch_port))
class TchRing:
def __init__(self):
self.fd = None
self.w = 0
def publish(self, fr33, fn):
if self.fd is None:
self.fd = os.open(TCH_DL_PATH, os.O_CREAT | os.O_RDWR, 0o644)
os.ftruncate(self.fd, 8 + TCH_DL_SLOTS * TCH_DL_SLOT)
os.pwrite(self.fd, (0).to_bytes(4, "little") + TCH_DL_SLOTS.to_bytes(4, "little"), 0)
self.w = 0
self.w += 1
rec = self.w.to_bytes(4, "little") + (fn & 0xffffffff).to_bytes(4, "little") + bytes(fr33)[:gsm.FR_BYTES]
rec += bytes(TCH_DL_SLOT - len(rec))
os.pwrite(self.fd, rec, 8 + ((self.w - 1) % TCH_DL_SLOTS) * TCH_DL_SLOT)
os.pwrite(self.fd, self.w.to_bytes(4, "little"), 0)
class Downlink:
def __init__(self, cfg, stats, cipher, dedicated, tch, feeder, record, timeslots):
self.cfg = cfg
self.stats = stats
self.cipher = cipher
self.dedicated = dedicated
self.tch = tch
self.feed = feeder
self.record = record
self.timeslots = timeslots
self.blocks = {}
self.ring = TchRing()
self.tch_acc = []
self.tch_last_fn = None
self.sacch_acc = {}
self._deciphered = {}
tch.on_close.append(self._tch_reset)
def _tch_reset(self):
self.tch_acc = []
self.tch_last_fn = None
self.sacch_acc = {}
self._deciphered = {}
def dispatch(self, tn, fn, burst):
if self.record:
self.record.feed_dl(fn, tn, burst)
if tn == 0 and fn % 51 in gsm.SCH_SLOTS_51:
if gsm.sch_decode(burst):
self.feed.sch(fn)
return
if tn == self.tch.active_tn():
self._tch(tn, fn, burst)
return
phys = self.timeslots.get(tn, "")
plan = gsm.plan_for(phys)
if plan is not None:
self._signalling(tn, fn, burst, plan)
def _active_subchannel(self, tn, plan):
ded = self.dedicated.plan()
if ded is None:
return None, False
dplan, ss, dtn = ded
if dtn != tn:
return None, True
if dplan is not plan:
return None, False
return ss % plan.n_sub, False
def _signalling(self, tn, fn, burst, plan):
m51 = fn % 51
base = next((b for b in plan.block_bases if b <= m51 <= b + 3), None)
if base is None:
return
fn0 = fn - (m51 - base)
ss, elsewhere = self._active_subchannel(tn, plan)
if elsewhere and base in plan.ded_bases:
self.stats.dl_skipped += 1
return
if ss is not None:
if base in plan.sdcch_dl and base != plan.sdcch_dl[ss]:
return
if base in plan.sacch_dl and (fn0 % 102) != plan.sacch_dl102[ss]:
return
if len(self.blocks) > 64:
self.blocks.clear()
if gsm.is_dummy(burst):
self.stats.dl_dummy += 1
return
acc = self.blocks.setdefault((tn, fn0), [])
acc.append((fn, burst))
self.stats.dl_bursts += 1
if len(acc) < 4:
return
del self.blocks[(tn, fn0)]
l2 = self._decode(acc, base in plan.ded_bases)
self.stats.block(tn, base, l2 is not None)
if l2 is None:
self.stats.dl_crc_fail += 1
return
self.stats.dl_blocks += 1
if base in plan.sdcch_dl:
self.feed.l2(fn0, gsm.GSMTAP_SDCCH4, l2, tn)
self._rr_downlink(l2, fn0)
return
if base in plan.sacch_dl:
self.feed.l2(fn0, gsm.GSMTAP_SACCH, l2, tn)
return
mt = None
if len(l2) >= 3 and l2[1] == gsm.RR_PD:
mt = l2[2]
elif len(l2) >= 5 and l2[3] == gsm.RR_PD:
mt = l2[4]
if mt in (gsm.RR_IMMEDIATE_ASSIGNMENT, gsm.RR_IMMEDIATE_ASSIGNMENT_EXT):
self.cipher.release("IMMEDIATE ASSIGNMENT")
if mt in gsm.SI_TYPES:
self.feed.l2(fn0, gsm.GSMTAP_BCCH, l2, tn)
elif mt in gsm.CCCH_TYPES:
self.feed.l2(fn0, gsm.GSMTAP_CCCH, l2, tn)
def _clear(self, bursts):
return [gsm.coded_from_burst(b) for _, b in bursts]
def _ciphered(self, bursts):
if len(self._deciphered) > 64:
self._deciphered.clear()
out = []
for fn, b in bursts:
d = self._deciphered.get(fn)
if d is None:
d = self._deciphered[fn] = gsm.coded_from_burst(self.cipher.apply(b, fn, False))
out.append(d)
return out
def _decode(self, bursts, dedicated):
# Comme dans un vrai reseau : clair, puis chiffre. La BTS n'active le
# chiffrement descendant qu'apres la premiere trame montante chiffree,
# donc SI5/SI6 et les premieres trames du SDCCH arrivent en clair meme
# quand le Kc est deja connu. On decode en clair tant que ca passe, et
# on verrouille l'etat global "descendant chiffre" des qu'un bloc ne
# decode qu'une fois dechiffre. Tous les canaux suivent cet etat.
if not dedicated or self.cipher.current() is None:
return gsm.xcch_decode(self._clear(bursts))
if not self.cipher.dl_active:
l2 = gsm.xcch_decode(self._clear(bursts))
if l2 is not None:
return l2
l2 = gsm.xcch_decode(self._ciphered(bursts))
if l2 is not None:
self.cipher.confirm_dl(bursts[0][0])
return l2
def _decode_tch(self, bursts):
# Meme regle pour le TCH : la parole suit l'etat global, une FACCH
# valide (CRC Fire) sert de preuve pour basculer l'etat.
if self.cipher.current() is None:
return gsm.tch_fr_decode(self._clear(bursts))
if self.cipher.dl_active:
return gsm.tch_fr_decode(self._ciphered(bursts))
rc, fr = gsm.tch_fr_decode(self._clear(bursts))
if rc == gsm.MACBLOCK_LEN:
return rc, fr
rc2, fr2 = gsm.tch_fr_decode(self._ciphered(bursts))
if rc2 == gsm.MACBLOCK_LEN:
self.cipher.confirm_dl(bursts[0][0])
return rc2, fr2
return rc, fr
def _rr_downlink(self, l2, fn):
off, mt = gsm.rr_message_type(l2)
if mt == gsm.RR_ASSIGNMENT_COMMAND and len(l2) > off + 4:
b0, b1, b2 = l2[off + 2], l2[off + 3], l2[off + 4]
if (b1 >> 4) & 1:
log.warning("ASSIGNMENT COMMAND avec saut de frequence : TCH non arme (fn=%u)", fn)
return
self.tch.arm(b0 & 0x07, (b1 >> 5) & 0x07, ((b1 & 0x03) << 8) | b2)
elif mt == gsm.RR_CHANNEL_RELEASE:
self.tch.release_requested("CHANNEL RELEASE")
def _tch(self, tn, fn, burst):
if gsm.is_dummy(burst):
self.stats.dl_dummy += 1
return
if fn % 26 == gsm.sacch_tf_frame(tn):
self._tch_sacch(tn, fn, burst)
return
if not gsm.is_tch_carrier(fn):
return
if self.tch_last_fn is not None and fn != gsm.next_tch_carrier(self.tch_last_fn):
self.tch_acc = []
self.tch_last_fn = fn
self.tch_acc.append((fn, burst))
if len(self.tch_acc) > 8:
del self.tch_acc[0]
if len(self.tch_acc) < 8 or gsm.tch_burst_index(fn) % 4 != 3:
return
rc, fr = self._decode_tch(self.tch_acc)
if rc == gsm.FR_BYTES:
self.stats.tch_dl += 1
self.ring.publish(fr, fn)
elif rc == gsm.MACBLOCK_LEN:
self.stats.tch_facch_dl += 1
l2 = fr[:gsm.MACBLOCK_LEN]
self.feed.l2(fn, gsm.GSMTAP_TCH_F, l2, tn)
if gsm.rr_message_type(l2)[1] == gsm.RR_CHANNEL_RELEASE:
self.tch.release_requested("CHANNEL RELEASE (FACCH)")
else:
self.stats.tch_crc += 1
def _tch_sacch(self, tn, fn, burst):
start = gsm.sacch_tf_block_base(tn)
block = fn - ((fn % 104 - start) % 104)
acc = self.sacch_acc.setdefault(block, [])
acc.append((fn, burst))
if len(self.sacch_acc) > 4:
for k in sorted(self.sacch_acc)[:-2]:
del self.sacch_acc[k]
if len(acc) < 4:
return
del self.sacch_acc[block]
l2 = self._decode(acc, True)
if l2 is None:
self.stats.tch_crc += 1
return
self.stats.tch_sacch_dl += 1
self.feed.l2(block, gsm.GSMTAP_TCH_ACCH, l2, tn)pont/gsm.py
pont/gsm.py
import ctypes
import struct
HYPERFRAME = 26 * 51 * 2048
FRAME_DUR = 60.0 / 13000.0
MACBLOCK_LEN = 23
FR_BYTES = 33
GSMTAP_BCCH = 0x01
GSMTAP_CCCH = 0x04
GSMTAP_SDCCH4 = 0x07
GSMTAP_TCH_F = 0x09
GSMTAP_ACCH = 0x80
GSMTAP_SACCH = GSMTAP_SDCCH4 | GSMTAP_ACCH
GSMTAP_TCH_ACCH = GSMTAP_TCH_F | GSMTAP_ACCH
RR_PD = 0x06
RR_ASSIGNMENT_COMMAND = 0x2e
RR_ASSIGNMENT_COMPLETE = 0x29
RR_CHANNEL_RELEASE = 0x0d
RR_IMMEDIATE_ASSIGNMENT = 0x3f
RR_IMMEDIATE_ASSIGNMENT_EXT = 0x39
SI_TYPES = frozenset((0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e))
CCCH_TYPES = frozenset((0x3f, 0x39, 0x3a, 0x21, 0x22, 0x24))
TSC7 = bytes((1, 1, 1, 0, 1, 1, 1, 1, 0, 0, 0, 1, 0, 0, 1, 0, 1, 1, 1, 0, 1, 1, 1, 1, 0, 0))
RACH_SYNC = bytes(int(c) for c in "01001011011111111001100110101010001111000")
RACH_SLOTS_51 = frozenset([4, 5] + list(range(14, 37)) + [45, 46])
SCH_SLOTS_51 = frozenset((1, 11, 21, 31, 41))
DUMMY_BURST = bytes((
0, 0, 0, 1, 1, 1, 1, 1, 0, 1, 1, 0, 1, 1, 1, 0, 1, 1, 0, 0, 0, 0, 0, 1, 0, 1, 0, 0, 1, 0, 0, 1, 1, 1, 0, 0, 0,
0, 0, 1, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 0, 0, 0, 1, 1, 1, 0, 0, 0, 1, 0, 1, 1, 1, 0,
0, 0, 1, 0, 1, 1, 1, 0, 0, 0, 1, 0, 1, 0, 1, 1, 1, 0, 1, 0, 0, 1, 0, 1, 0, 0, 0, 1, 1, 0, 0, 1, 1, 0, 0, 1, 1,
1, 0, 0, 1, 1, 1, 1, 0, 1, 0, 0, 1, 1, 1, 1, 1, 0, 0, 0, 1, 0, 0, 1, 0, 1, 1, 1, 1, 1, 0, 1, 0, 1, 0, 0, 0, 0))
DUMMY_TOLERANCE = 8
_SOFT = bytes([127, 129] + [129] * 254)
_ONEBIT = bytes([0] + [1] * 255)
ubit = ctypes.c_int8
sbit = ctypes.c_int8
_cod = ctypes.CDLL("libosmocoding.so", use_errno=True)
_cod.gsm0503_xcch_encode.argtypes = [ctypes.POINTER(ubit), ctypes.POINTER(ctypes.c_uint8)]
_cod.gsm0503_xcch_decode.argtypes = [ctypes.POINTER(ctypes.c_uint8), ctypes.POINTER(sbit),
ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int)]
_cod.gsm0503_rach_ext_encode.argtypes = [ctypes.POINTER(ubit), ctypes.c_uint16, ctypes.c_uint8, ctypes.c_bool]
_cod.gsm0503_sch_decode.argtypes = [ctypes.POINTER(ctypes.c_uint8), ctypes.POINTER(sbit)]
_cod.gsm0503_tch_fr_encode.argtypes = [ctypes.POINTER(ubit), ctypes.POINTER(ctypes.c_uint8), ctypes.c_int, ctypes.c_int]
_cod.gsm0503_tch_fr_decode.argtypes = [ctypes.POINTER(ctypes.c_uint8), ctypes.POINTER(sbit), ctypes.c_int,
ctypes.c_int, ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int)]
_gsm = ctypes.CDLL("libosmogsm.so", use_errno=True)
_gsm.osmo_a5.argtypes = [ctypes.c_int, ctypes.POINTER(ctypes.c_uint8), ctypes.c_uint32,
ctypes.POINTER(ubit), ctypes.POINTER(ubit)]
def normalize_bits(raw148):
return raw148.translate(_ONEBIT)
def is_dummy(burst):
errors = 0
for a, b in zip(burst, DUMMY_BURST):
if a != b:
errors += 1
if errors > DUMMY_TOLERANCE:
return False
return True
def coded_from_burst(burst):
return burst[3:61] + burst[87:145]
def burst_from_coded(coded):
return bytes(3) + bytes(coded[:58]) + TSC7 + bytes(coded[58:116]) + bytes(3)
def _soft(bursts):
return b"".join(bytes(b).translate(_SOFT) for b in bursts)
def xcch_decode(bursts4):
buf = (sbit * (4 * 116)).from_buffer_copy(_soft(bursts4))
l2 = (ctypes.c_uint8 * MACBLOCK_LEN)()
ne = ctypes.c_int()
nb = ctypes.c_int()
rc = _cod.gsm0503_xcch_decode(l2, buf, ctypes.byref(ne), ctypes.byref(nb))
return bytes(l2) if rc == 0 else None
def xcch_encode(l2):
data = (ctypes.c_uint8 * MACBLOCK_LEN)(*l2[:MACBLOCK_LEN])
e = (ubit * (4 * 116))()
_cod.gsm0503_xcch_encode(e, data)
raw = bytes(e)
return [burst_from_coded(raw[k * 116:(k + 1) * 116]) for k in range(4)]
def tch_fr_decode(bursts8):
buf = (sbit * (8 * 116)).from_buffer_copy(_soft(bursts8))
out = (ctypes.c_uint8 * FR_BYTES)()
ne = ctypes.c_int()
nb = ctypes.c_int()
rc = _cod.gsm0503_tch_fr_decode(out, buf, 1, 0, ctypes.byref(ne), ctypes.byref(nb))
return rc, bytes(out)
def tch_fr_encode_into(buf24, payload):
if payload:
data = (ctypes.c_uint8 * len(payload))(*payload)
_cod.gsm0503_tch_fr_encode(buf24, data, len(payload), 1)
else:
_cod.gsm0503_tch_fr_encode(buf24, None, 0, 1)
def sch_decode(burst):
d78 = burst[3:42] + burst[106:145]
buf = (sbit * 78)(*[-127 if b else 127 for b in d78])
sb = (ctypes.c_uint8 * 4)()
return _cod.gsm0503_sch_decode(sb, buf) == 0
def rach_burst(ra, bsic):
coded = (ubit * 40)()
_cod.gsm0503_rach_ext_encode(coded, ctypes.c_uint16(ra), ctypes.c_uint8(bsic), False)
return bytes(8) + RACH_SYNC + bytes(coded)[:36] + bytes(148 - 8 - 41 - 36)
def a5_keystream(algo, kc, fn):
key = (ctypes.c_uint8 * 8)(*kc)
dl = (ubit * 114)()
ul = (ubit * 114)()
_gsm.osmo_a5(algo, key, ctypes.c_uint32(fn & 0xffffffff), dl, ul)
return bytes(dl), bytes(ul)
def a5_xor(burst, ks):
b = bytearray(burst)
for i in range(57):
b[3 + i] ^= ks[i]
b[88 + i] ^= ks[57 + i]
return bytes(b)
def gsmtap_header(arfcn, fn, chan, tn, uplink):
a = arfcn | (0x4000 if uplink else 0)
return struct.pack(">BBBBHbBIBBBB", 2, 4, 0x01, tn, a, 0, 0, fn, chan, 0, 0, 0)
def rr_message_type(l2):
for off in (0, 3):
if len(l2) > off + 1 and l2[off] == RR_PD:
if off == 3 and (l2[2] >> 2) < 2:
continue
return off, l2[off + 1]
return None, None
def tch_burst_index(fn):
m = fn % 26
return m if m < 12 else m - 1
def is_tch_carrier(fn):
return fn % 26 not in (12, 25)
def next_tch_carrier(fn):
n = (fn + 1) % HYPERFRAME
while not is_tch_carrier(n):
n = (n + 1) % HYPERFRAME
return n
def sacch_tf_frame(tn):
return 12 if tn % 2 == 0 else 25
def sacch_tf_block_base(tn):
# 45.002 table 5 : le bloc SACCH/TF du TN commence a la trame 12 + 13*TN de la 104-multitrame
return (12 + 13 * tn) % 104
def sacch_tf_block_start(fn, tn):
return fn % 104 == sacch_tf_block_base(tn)
class SlotPlan:
def __init__(self, name, sdcch_dl, sacch_dl, other_dl):
self.name = name
self.n_sub = len(sdcch_dl)
self.sdcch_dl = tuple(sdcch_dl)
self.sacch_dl = tuple(sacch_dl)
half = len(sacch_dl)
self.sacch_dl102 = tuple(sacch_dl[i % half] + (51 if i >= half else 0) for i in range(self.n_sub))
self.ded_bases = tuple(sorted(set(self.sdcch_dl + self.sacch_dl)))
self.block_bases = tuple(sorted(set(self.sdcch_dl + self.sacch_dl + tuple(other_dl))))
def ul_base(self, ss):
return (self.sdcch_dl[ss % self.n_sub] + 15) % 51
PLAN_SDCCH4 = SlotPlan("CCCH+SDCCH4", (22, 26, 32, 36), (42, 46), (2, 6, 12, 16))
PLAN_SDCCH8 = SlotPlan("SDCCH8", (0, 4, 8, 12, 16, 20, 24, 28), (32, 36, 40, 44), ())
def plan_for(phys):
if phys.startswith("SDCCH8"):
return PLAN_SDCCH8
if phys.startswith("CCCH"):
return PLAN_SDCCH4
return None
def load_timeslots(path):
ts = {}
try:
cur = None
n_bts = 0
for line in open(path, encoding="utf-8", errors="ignore"):
s = line.strip()
if s.startswith("bts "):
n_bts += 1
if n_bts > 1:
break
cur = None
elif s.startswith("timeslot "):
cur = int(s.split()[1])
elif s.startswith("phys_chan_config") and cur is not None:
ts[cur] = s.split(None, 1)[1].strip()
cur = None
except OSError:
pass
return ts or {0: "CCCH+SDCCH4", 1: "SDCCH8", 2: "TCH/F", 3: "TCH/F"}pont/__init__.py
pont/__init__.py
import gc
import logging
import signal
import sys
import threading
import time
from . import config as _config
from .cipher import Cipher
from .downlink import Downlink, Feeder
from .gsm import load_timeslots, plan_for
from .record import Recorder
from .state import Dedicated, Tch
from .stats import Reporter, Stats
from .trx import Clock, Transmitter, Trx
from .uplink import TchScheduler, Uplink
log = logging.getLogger("pont")
def _guarded(name, fn):
def run():
while True:
try:
fn()
log.warning("thread %s termine", name)
return
except Exception:
log.exception("thread %s : exception, relance dans 1 s", name)
time.sleep(1)
return threading.Thread(target=run, name=name, daemon=True)
def main(argv=None):
cfg = _config.parse(argv)
logging.basicConfig(stream=sys.stdout, level=logging.INFO,
format="%(asctime)s [pont] %(message)s", datefmt="%H:%M:%S")
gc.disable()
gc.freeze()
stats = Stats()
clock = Clock()
cipher = Cipher(cfg.kc_retention, stats)
dedicated = Dedicated()
tch = Tch(stats)
record = Recorder(cfg, clock) if cfg.record else None
trx = Trx(cfg, clock, stats, cipher, record)
transmitter = Transmitter(cfg, clock, trx, stats)
feeder = Feeder(cfg, stats)
timeslots = load_timeslots(cfg.bsc_cfg)
downlink = Downlink(cfg, stats, cipher, dedicated, tch, feeder, record, timeslots)
uplink = Uplink(cfg, clock, stats, dedicated, tch, transmitter, feeder, cipher)
scheduler = TchScheduler(cfg, clock, stats, tch, uplink, transmitter)
log.info("pont TRX : ports %d/%d/%d, ARFCN %d, BSIC %d, avance UL %d trames",
cfg.trx_base, cfg.trx_base + 1, cfg.trx_base + 2, cfg.arfcn, cfg.bsic, cfg.ul_fn_advance)
for tn in sorted(timeslots):
plan = plan_for(timeslots[tn])
log.info("TS%d %-12s %s", tn, timeslots[tn], ("plan %s, %d sous-voies" % (plan.name, plan.n_sub)) if plan else "")
log.info("QEMU : GSMTAP udp/%d, SCH udp/%d ; tap GSMTAP %s", cfg.gsmtap_port, cfg.sch_port,
("udp/%d" % cfg.tap_port) if cfg.tap else "coupe")
threads = [
_guarded("ctrl", trx.run_ctrl),
_guarded("clock", trx.run_clock),
_guarded("data", lambda: trx.run_data(downlink.dispatch)),
transmitter, uplink, scheduler,
Reporter(stats, clock, dedicated, record),
]
if record:
signal.signal(signal.SIGUSR1, record.split)
threads.append(record)
for t in threads:
t.start()
while True:
time.sleep(3600)pont/pont.py
pont/record.py
pont/record.py
import logging
import os
import threading
import time
import numpy as np
from . import gsm
log = logging.getLogger("pont")
SAMPLES_PER_FRAME = 1250
SAMPLES_PER_SLOT_X4 = 625
class Direction:
def __init__(self, name, path, fifo_path, max_bytes, min_free, osr, taps):
self.name = name
self.path = path
self.fifo_path = fifo_path
self.max_bytes = max_bytes
self.min_free = min_free
self.osr = osr
self.taps = taps
self.lock = threading.Lock()
self.frames = {}
self.fh = None
self.bytes = 0
self.off = False
self.restart = False
self.fifo_fd = None
self.fifo_pending = b""
self.fifo_retry = 0
self.n_frames = 0
self.n_bursts = 0
self.n_fifo_drop = 0
def feed(self, fn, tn, burst):
if self.off:
return
with self.lock:
self.frames.setdefault(fn, {})[tn] = burst
def open(self):
if self.fh is not None:
self.fh.close()
self.fh = None
try:
os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
st = os.statvfs(os.path.dirname(self.path) or ".")
if st.f_bavail * st.f_frsize < self.min_free:
log.warning("enregistrement %s coupe : moins de %d Mo libres", self.name, self.min_free // 1048576)
return False
self.fh = open(self.path, "wb", buffering=1024 * 1024)
except OSError as e:
log.warning("enregistrement %s impossible (%s)", self.name, e)
return False
self.bytes = 0
log.info("enregistrement %s -> %s (buffer roulant de %d Mo)", self.name, self.path, self.max_bytes // 1048576)
return True
def modulate(self, burst):
nrz = 2.0 * np.frombuffer(bytes(burst), dtype=np.uint8).astype(np.float64) - 1.0
prev = np.concatenate(([nrz[0]], nrz[:-1]))
up = np.zeros(len(nrz) * self.osr)
up[::self.osr] = nrz * prev
phase = np.cumsum(np.convolve(up, self.taps, mode="same")) * (np.pi / 2.0)
return np.exp(1j * phase).astype(np.complex64)
def _fifo(self, octets):
if not self.fifo_path:
return
if self.fifo_fd is None:
if self.n_frames < self.fifo_retry:
return
self.fifo_retry = self.n_frames + 217
try:
if not os.path.exists(self.fifo_path):
os.mkfifo(self.fifo_path, 0o644)
self.fifo_fd = os.open(self.fifo_path, os.O_WRONLY | os.O_NONBLOCK)
except OSError:
return
data = self.fifo_pending + octets
maxq = 3 * len(octets)
if len(data) > maxq:
data = data[len(data) - maxq:]
self.n_fifo_drop += 1
try:
n = os.write(self.fifo_fd, data)
self.fifo_pending = data[n:]
except BlockingIOError:
self.fifo_pending = data
except OSError:
os.close(self.fifo_fd)
self.fifo_fd = None
self.fifo_pending = b""
def write_frame(self, fn, spf, step, blen, silence):
if self.off or self.fh is None:
return
with self.lock:
slots = self.frames.pop(fn, None)
if len(self.frames) > 512:
self.frames.clear()
if slots:
frame = np.zeros(spf, dtype=np.complex64)
for tn, burst in slots.items():
o = tn * step
frame[o:o + blen] = self.modulate(burst)
self.n_bursts += 1
else:
frame = silence
octets = frame.tobytes()
self._fifo(octets)
try:
self.fh.write(octets)
except OSError as e:
log.warning("enregistrement %s : ecriture echouee (%s), coupe", self.name, e)
self.off = True
return
self.bytes += len(octets)
self.n_frames += 1
if self.restart or self.bytes >= self.max_bytes:
self.restart = False
self.fh.flush()
self.fh.seek(0)
self.fh.truncate(0)
self.bytes = 0
class Recorder(threading.Thread):
def __init__(self, cfg, clock):
super().__init__(name="record", daemon=True)
self.cfg = cfg
self.clock = clock
osr = cfg.record_osr
n = 4 * osr
t = np.arange(-n, n + 1) / float(osr)
g = np.exp(-2.0 * (np.pi ** 2) * (0.3 ** 2) * t ** 2 / np.log(2.0))
taps = g / g.sum()
max_bytes = cfg.record_max_mb * 1024 * 1024
min_free = cfg.record_min_free_mb * 1024 * 1024
self.dl = Direction("descendant", cfg.record_dl_path, cfg.record_dl_fifo, max_bytes, min_free, osr, taps)
self.ul = Direction("montant", cfg.record_ul_path, cfg.record_ul_fifo, max_bytes, min_free, osr, taps)
self.ul.off = not cfg.record_ul
def feed_dl(self, fn, tn, burst):
self.dl.feed(fn, tn, burst)
def feed_ul(self, fn, tn, burst):
self.ul.feed(fn, tn, burst)
def split(self, *_):
self.dl.restart = True
self.ul.restart = True
def summary(self):
return " | IQ dl=%dtr/%dbu ul=%dtr/%dbu fifo_drop=%d/%d" % (
self.dl.n_frames, self.dl.n_bursts, self.ul.n_frames, self.ul.n_bursts,
self.dl.n_fifo_drop, self.ul.n_fifo_drop)
def run(self):
osr = self.cfg.record_osr
if osr % 4:
log.warning("enregistrement refuse : OSR=%d doit etre multiple de 4", osr)
return
if not self.dl.open():
self.dl.off = True
if not self.ul.off and not self.ul.open():
self.ul.off = True
if self.dl.off and self.ul.off:
return
spf = SAMPLES_PER_FRAME * osr
step = (SAMPLES_PER_SLOT_X4 * osr) // 4
blen = 148 * osr
silence = np.zeros(spf, dtype=np.complex64)
nxt = (self.clock.fn() + 2) % gsm.HYPERFRAME
while True:
d = (self.clock.fn() - nxt) % gsm.HYPERFRAME
if d > gsm.HYPERFRAME // 2 or d < 2:
time.sleep(gsm.FRAME_DUR / 2.0)
continue
if d > 650:
with self.dl.lock:
self.dl.frames.clear()
with self.ul.lock:
self.ul.frames.clear()
nxt = (self.clock.fn() + 2) % gsm.HYPERFRAME
continue
self.dl.write_frame(nxt, spf, step, blen, silence)
self.ul.write_frame(nxt, spf, step, blen, silence)
nxt = (nxt + 1) % gsm.HYPERFRAMEpont/state.py
pont/state.py
import logging
import os
import struct
import threading
import time
from . import gsm
log = logging.getLogger("pont")
DCCH_CFG = "/dev/shm/calypso_dcch_cfg"
TCH_CFG = "/dev/shm/calypso_tch_cfg"
DCCH_TTL = 0.1
DCCH_RELEASED = 0xFF
class Dedicated:
def __init__(self):
self._fd = None
self._next = 0.0
self._value = None
self._seq = 0
def read(self):
now = time.monotonic()
if now < self._next:
return self._value
self._next = now + DCCH_TTL
try:
if self._fd is None:
self._fd = os.open(DCCH_CFG, os.O_RDONLY)
b = os.pread(self._fd, 16, 0)
except OSError:
if self._fd is not None:
os.close(self._fd)
self._fd = None
return self._value
if len(b) < 8:
return self._value
seq = struct.unpack_from("<I", b, 0)[0]
if not seq or seq == self._seq:
return self._value
self._seq = seq
self._value = None if b[4] == DCCH_RELEASED else (b[4] & 1, b[5] & 7, b[6] & 7)
return self._value
def plan(self):
v = self.read()
if v is None:
return None
kind, ss, tn = v
return (gsm.PLAN_SDCCH8 if kind else gsm.PLAN_SDCCH4), ss, tn
class Tch:
def __init__(self, stats):
self.stats = stats
self.lock = threading.Lock()
self.tn = None
self.tsc = 0
self.open = False
self.seq = 0
self.on_close = []
self.release_seen = None
def arm(self, tn, tsc, arfcn):
with self.lock:
self.seq += 1
self.tn = tn
self.tsc = tsc
self.open = False
self._write_cfg(tn, tsc, arfcn, self.seq)
log.info("ASSIGNMENT COMMAND : TCH TN=%d TSC=%d ARFCN=%d arme, en attente du mobile", tn, tsc, arfcn)
def prove(self, source):
with self.lock:
if self.tn is None or self.open:
return
self.open = True
tn = self.tn
log.info("le mobile est sur le TCH TN=%d (%s) : montant TCH ouvert", tn, source)
def release_requested(self, reason):
with self.lock:
if self.release_seen is not None:
return
self.release_seen = time.monotonic()
tn = self.tn
if tn is None:
log.info("%s : en attente de la liberation du canal par le mobile", reason)
else:
log.info("%s : TCH TN=%d maintenu jusqu'a la liberation par le mobile", reason, tn)
def release_overdue(self, delay):
with self.lock:
return self.release_seen is not None and time.monotonic() - self.release_seen > delay
def close(self, reason):
with self.lock:
self.release_seen = None
if self.tn is None:
return
tn = self.tn
self.tn = None
self.open = False
self._write_cfg(0, 0, 0, 0)
log.info("TCH TN=%d ferme (%s)", tn, reason)
for cb in self.on_close:
cb()
def active_tn(self):
return self.tn
def is_open(self):
return self.open and self.tn is not None
@staticmethod
def _write_cfg(tn, tsc, arfcn, seq):
buf = bytearray(16)
buf[4] = tn & 0xff
buf[5] = tsc & 0xff
buf[6:8] = int(arfcn).to_bytes(2, "little")
buf[0:4] = int(seq).to_bytes(4, "little")
fd = os.open(TCH_CFG, os.O_CREAT | os.O_WRONLY, 0o644)
try:
os.pwrite(fd, bytes(buf), 0)
finally:
os.close(fd)pont/stats.py
pont/stats.py
import logging
import threading
import time
log = logging.getLogger("pont")
class Stats:
def __init__(self):
self.dl_bursts = 0
self.dl_blocks = 0
self.dl_crc_fail = 0
self.dl_dummy = 0
self.dl_skipped = 0
self.ul_sent = 0
self.ul_late = 0
self.rach = 0
self.tch_dl = 0
self.tch_facch_dl = 0
self.tch_sacch_dl = 0
self.tch_crc = 0
self.tch_ul = 0
self.tch_ul_bursts = 0
self.tch_ul_dropped = 0
self.facch_ul = 0
self.sacch_ul = 0
self.a5_dl = 0
self.a5_ul = 0
self.by_base = {}
def block(self, tn, base, ok):
st = self.by_base.setdefault((tn, base), [0, 0])
st[0 if ok else 1] += 1
class Reporter(threading.Thread):
def __init__(self, stats, clock, dedicated, record, period=5.0):
super().__init__(name="stats", daemon=True)
self.stats = stats
self.clock = clock
self.dedicated = dedicated
self.record = record
self.period = period
def run(self):
s = self.stats
while True:
time.sleep(self.period)
log.info("STATS fn=%u | DL bursts=%d blocs=%d crc=%d dummy=%d hors_voie=%d | UL bursts=%d tard=%d rach=%d"
" | TCH dl=%d facch=%d sacch=%d crc=%d ul=%d bursts=%d perdus=%d | FACCH ul=%d SACCH ul=%d"
" | A5 dl=%d ul=%d%s",
self.clock.fn(), s.dl_bursts, s.dl_blocks, s.dl_crc_fail, s.dl_dummy, s.dl_skipped,
s.ul_sent, s.ul_late, s.rach, s.tch_dl, s.tch_facch_dl, s.tch_sacch_dl, s.tch_crc,
s.tch_ul, s.tch_ul_bursts, s.tch_ul_dropped, s.facch_ul, s.sacch_ul, s.a5_dl, s.a5_ul,
self.record.summary() if self.record else "")
ded = self.dedicated.plan()
if ded:
plan, ss, tn = ded
log.info("canal dedie : %s SS=%d TS=%d (DL fn%%51=%d, UL fn%%51=%d)",
plan.name, ss, tn, plan.sdcch_dl[ss % plan.n_sub], plan.ul_base(ss))
if s.by_base:
log.info("blocs par TS/base51 (ok/echec) : %s",
" ".join("TS%d/%d:%d/%d" % (k[0], k[1], v[0], v[1]) for k, v in sorted(s.by_base.items())))pont/trx.py
pont/trx.py
import heapq
import logging
import socket
import struct
import threading
import time
from . import gsm
log = logging.getLogger("pont")
class Clock:
def __init__(self):
self.t0 = time.monotonic()
def fn(self):
return int((time.monotonic() - self.t0) / gsm.FRAME_DUR) % gsm.HYPERFRAME
def time_of(self, fn):
cur = self.fn()
delta = (fn - cur) % gsm.HYPERFRAME
if delta > gsm.HYPERFRAME // 2:
delta -= gsm.HYPERFRAME
return self.t0 + (cur + delta) * gsm.FRAME_DUR
def sleep_until_fn(self, fn):
dt = self.time_of(fn) - time.monotonic()
if dt > 0:
time.sleep(dt)
def _udp(bind_host, port):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind((bind_host, port))
return s
class Trx:
def __init__(self, cfg, clock, stats, cipher, record):
self.cfg = cfg
self.clock = clock
self.stats = stats
self.cipher = cipher
self.record = record
self.sk_clck = _udp(cfg.trx_bind, cfg.trx_base)
self.sk_ctrl = _udp(cfg.trx_bind, cfg.trx_base + 1)
self.sk_data = _udp(cfg.trx_bind, cfg.trx_base + 2)
self.bts_data = None
self.bts_clck = None
def run_ctrl(self):
while True:
data, addr = self.sk_ctrl.recvfrom(1500)
req = data.decode("latin1").strip("\0").strip()
if not req.startswith("CMD"):
continue
parts = req.split()
cmd = parts[1] if len(parts) > 1 else ""
args = (" " + " ".join(parts[2:])) if len(parts) > 2 else ""
self.sk_ctrl.sendto(("RSP %s 0%s" % (cmd, args)).encode() + b"\0", addr)
if self.bts_clck is None:
self.bts_clck = (addr[0], self.cfg.trx_base + 100)
log.info("BTS %s : horloge armee vers le port %d", addr[0], self.bts_clck[1])
if cmd not in ("SETPOWER", "NOHANDOVER"):
log.info("CTRL %s%s", cmd, args)
def run_clock(self):
while True:
if self.bts_clck:
self.sk_clck.sendto(("IND CLOCK %u" % self.clock.fn()).encode() + b"\0", self.bts_clck)
time.sleep(gsm.FRAME_DUR * self.cfg.clock_period)
def run_data(self, on_burst):
while True:
data, addr = self.sk_data.recvfrom(2000)
self.bts_data = addr
if len(data) < 6 + 148:
continue
tn = data[0] & 0x07
fn = struct.unpack_from(">L", data, 1)[0]
on_burst(tn, fn, gsm.normalize_bits(data[6:154]))
def send_ul(self, tn, fn, burst, cipher):
if self.bts_data is None:
return
if cipher:
burst = self.cipher.apply(burst, fn, True)
hdr = bytes([tn & 0x07]) + struct.pack(">L", fn) + bytes([self.cfg.rssi & 0xFF]) + struct.pack(">h", 0)
self.sk_data.sendto(hdr + bytes(255 if b else 0 for b in burst), self.bts_data)
self.stats.ul_sent += 1
if self.record:
self.record.feed_ul(fn, tn, burst)
class Transmitter(threading.Thread):
def __init__(self, cfg, clock, trx, stats):
super().__init__(name="tx", daemon=True)
self.cfg = cfg
self.clock = clock
self.trx = trx
self.stats = stats
self.cond = threading.Condition()
self.heap = []
self.seq = 0
def schedule(self, tn, fn_air, burst, cipher):
post = self.clock.time_of((fn_air - self.cfg.ul_fn_advance) % gsm.HYPERFRAME)
with self.cond:
self.seq += 1
heapq.heappush(self.heap, (post, self.seq, tn, fn_air, burst, cipher))
self.cond.notify()
def run(self):
while True:
with self.cond:
while not self.heap:
self.cond.wait()
post = self.heap[0][0]
dt = post - time.monotonic()
if dt > 0:
self.cond.wait(dt)
continue
_, _, tn, fn_air, burst, cipher = heapq.heappop(self.heap)
cur = self.clock.fn()
off = (fn_air - (cur + self.cfg.ul_fn_advance) + gsm.HYPERFRAME // 2) % gsm.HYPERFRAME - gsm.HYPERFRAME // 2
if abs(off) > self.cfg.window_tol:
self.stats.ul_late += 1
continue
self.trx.send_ul(tn, fn_air, burst, cipher)pont/uplink.py
pont/uplink.py
import collections
import ctypes
import logging
import os
import struct
import threading
import time
from . import gsm
log = logging.getLogger("pont")
SB_RACH = "/dev/shm/calypso_rach"
SB_SDCCH_UL = "/dev/shm/calypso_sdcch_ul"
SB_FACCH_UL = "/dev/shm/calypso_tch_facch_ul"
SB_SACCH_UL = "/dev/shm/calypso_tch_sacch_ul"
SB_TCH_UL = "/dev/shm/calypso_tch_ul"
TCH_UL_SLOT = 64
TCH_UL_FR_OFS = 16
POLL = gsm.FRAME_DUR / 8
TX_BURSTS = 24
BPLEN = 116
RELEASE_FALLBACK = 20.0
class Sideband:
def __init__(self, path, size):
self.path = path
self.size = size
self.fd = None
self.seq = 0
def read(self):
try:
if self.fd is None:
self.fd = os.open(self.path, os.O_RDONLY)
return os.pread(self.fd, self.size, 0)
except OSError:
if self.fd is not None:
os.close(self.fd)
self.fd = None
return b""
def new_record(self):
b = self.read()
if len(b) < self.size:
return None
seq = struct.unpack_from("<I", b, 0)[0]
if not seq or seq == self.seq:
return None
self.seq = seq
return b
def skip_pending(self):
b = self.read()
if len(b) >= 4:
self.seq = struct.unpack_from("<I", b, 0)[0]
class TchUplinkRing:
def __init__(self):
self.fd = None
self.last = 0
def drain(self):
try:
if self.fd is None:
self.fd = os.open(SB_TCH_UL, os.O_RDONLY)
hdr = os.pread(self.fd, 8, 0)
except OSError:
if self.fd is not None:
os.close(self.fd)
self.fd = None
return [], 0
if len(hdr) < 8:
return [], 0
w, n = struct.unpack("<II", hdr)
if not w or not n:
return [], 0
lost = 0
if self.last == 0:
self.last = w
elif (w - self.last) > n:
lost = (w - self.last) - n
self.last = w - n
frames = []
while self.last < w:
self.last += 1
sl = os.pread(self.fd, TCH_UL_SLOT, 8 + ((self.last - 1) % n) * TCH_UL_SLOT)
if len(sl) >= TCH_UL_FR_OFS + gsm.FR_BYTES:
frames.append(sl[TCH_UL_FR_OFS:TCH_UL_FR_OFS + gsm.FR_BYTES])
return frames, lost
class Uplink(threading.Thread):
def __init__(self, cfg, clock, stats, dedicated, tch, transmitter, feeder, cipher):
super().__init__(name="uplink", daemon=True)
self.cfg = cfg
self.clock = clock
self.stats = stats
self.dedicated = dedicated
self.tch = tch
self.cipher = cipher
self.ded_active = False
self.tx = transmitter
self.feed = feeder
self.sb_rach = Sideband(SB_RACH, 12)
self.sb_sdcch = Sideband(SB_SDCCH_UL, 39)
self.sb_facch = Sideband(SB_FACCH_UL, 39)
self.sb_sacch = Sideband(SB_SACCH_UL, 39)
self.ring = TchUplinkRing()
self.q_lock = threading.Lock()
self.q_facch = collections.deque(maxlen=8)
self.q_voice = collections.deque(maxlen=4)
self.sacch_l2 = None
self.tch_epoch = -1
for sb in (self.sb_rach, self.sb_sdcch, self.sb_facch, self.sb_sacch):
sb.skip_pending()
def _next_fn(self, min_advance, accept):
start = self.clock.fn() + min_advance
for k in range(104):
fn = (start + k) % gsm.HYPERFRAME
if accept(fn):
return fn
return start % gsm.HYPERFRAME
def _queue_facch(self, l2):
with self.q_lock:
self.q_facch.append(bytes(l2[:gsm.MACBLOCK_LEN]))
self.stats.facch_ul += 1
self.feed.l2(self.clock.fn(), gsm.GSMTAP_TCH_F, l2[:gsm.MACBLOCK_LEN], self.tch.active_tn() or 0, True)
def _poll_rach(self):
b = self.sb_rach.new_record()
if b is None:
return
_, ra, bsic, fn = struct.unpack_from("<IBBxxI", b, 0)
burst = gsm.rach_burst(ra, bsic)
self.tx.schedule(0, self._next_fn(4, lambda f: f % 51 in gsm.RACH_SLOTS_51), burst, False)
self.stats.rach += 1
def _poll_sdcch(self):
b = self.sb_sdcch.new_record()
if b is None:
return
l2 = b[16:39]
tn_tch = self.tch.active_tn()
if tn_tch is not None and not self.tch.is_open():
if gsm.rr_message_type(l2)[1] == gsm.RR_ASSIGNMENT_COMPLETE:
self.tch.prove("ASSIGNMENT COMPLETE")
self._queue_facch(l2)
return
elif self.tch.is_open():
self._queue_facch(l2)
return
ded = self.dedicated.plan()
if ded is None:
return
plan, ss, tn = ded
self.feed.l2(self.clock.fn(), gsm.GSMTAP_SDCCH4, l2, tn, True)
base = plan.ul_base(ss)
fn0 = self._next_fn(4, lambda f: f % 51 == base)
for j, burst in enumerate(gsm.xcch_encode(l2)):
self.tx.schedule(tn, (fn0 + j) % gsm.HYPERFRAME, burst, True)
def _poll_facch(self):
tn = self.tch.active_tn()
if tn is None:
self.tch_epoch = -1
return
if self.tch_epoch != self.tch.seq:
self.tch_epoch = self.tch.seq
self.sb_facch.skip_pending()
self.sb_sacch.skip_pending()
return
b = self.sb_facch.new_record()
if b is None:
return
l2 = b[16:39]
self.tch.prove("FACCH montante%s" % (", ASSIGNMENT COMPLETE" if gsm.rr_message_type(l2)[1] == gsm.RR_ASSIGNMENT_COMPLETE else ""))
self._queue_facch(l2)
def _poll_sacch(self):
b = self.sb_sacch.new_record()
if b is None:
return
self.sacch_l2 = bytes(b[16:39])
self.stats.sacch_ul += 1
def _poll_release(self):
active = self.dedicated.plan() is not None
if self.ded_active and not active:
reason = "canal libere par le mobile"
elif self.tch.release_overdue(RELEASE_FALLBACK):
reason = "CHANNEL RELEASE sans liberation du mobile depuis %ds" % RELEASE_FALLBACK
else:
self.ded_active = active
return
self.ded_active = active
self.tch.close(reason)
self.cipher.release(reason)
def _poll_voice(self):
if not self.tch.is_open():
return
frames, lost = self.ring.drain()
self.stats.tch_ul_dropped += lost
for fr in frames:
with self.q_lock:
if len(self.q_voice) == self.q_voice.maxlen:
self.stats.tch_ul_dropped += 1
self.q_voice.append(bytes(fr))
self.stats.tch_ul += 1
def run(self):
while True:
self._poll_release()
self._poll_rach()
self._poll_sdcch()
self._poll_facch()
self._poll_sacch()
self._poll_voice()
time.sleep(POLL)
class TchScheduler(threading.Thread):
def __init__(self, cfg, clock, stats, tch, uplink, transmitter):
super().__init__(name="tch-tx", daemon=True)
self.cfg = cfg
self.clock = clock
self.stats = stats
self.tch = tch
self.uplink = uplink
self.tx = transmitter
self.buf = (gsm.ubit * (TX_BURSTS * BPLEN))()
self.mask = 0
self.sent_sacch_block = None
tch.on_close.append(self._reset)
def _reset(self):
ctypes.memset(self.buf, 0, TX_BURSTS * BPLEN)
self.mask = 0
self.sent_sacch_block = None
with self.uplink.q_lock:
self.uplink.q_facch.clear()
self.uplink.q_voice.clear()
def _load_block(self):
ctypes.memmove(self.buf, ctypes.byref(self.buf, 4 * BPLEN), 20 * BPLEN)
ctypes.memset(ctypes.byref(self.buf, 20 * BPLEN), 0, 4 * BPLEN)
self.mask = (self.mask << 4) & 0xFFFFFFFF
with self.uplink.q_lock:
facch = self.uplink.q_facch.popleft() if self.uplink.q_facch else None
voice = self.uplink.q_voice.popleft() if self.uplink.q_voice else None
gsm.tch_fr_encode_into(self.buf, facch if facch is not None else voice)
def _sacch(self, tn, fn_air):
if not gsm.sacch_tf_block_start(fn_air, tn):
return
block = fn_air - (fn_air % 104)
if self.uplink.sacch_l2 is None or self.sent_sacch_block == block:
return
self.sent_sacch_block = block
for j, burst in enumerate(gsm.xcch_encode(self.uplink.sacch_l2)):
self.tx.schedule(tn, (fn_air + 26 * j) % gsm.HYPERFRAME, burst, True)
def run(self):
fn_air = None
while True:
tn = self.tch.active_tn()
if tn is None or not self.tch.is_open():
fn_air = None
time.sleep(0.05)
continue
if fn_air is None:
fn_air = (self.clock.fn() + self.cfg.ul_fn_advance + 1) % gsm.HYPERFRAME
else:
fn_air = (fn_air + 1) % gsm.HYPERFRAME
self.clock.sleep_until_fn((fn_air - self.cfg.ul_fn_advance) % gsm.HYPERFRAME)
self._sacch(tn, fn_air)
if not gsm.is_tch_carrier(fn_air):
continue
bid = gsm.tch_burst_index(fn_air) % 4
if bid == 0:
self._load_block()
elif not (self.mask & 0x01):
continue
base = bid * BPLEN
burst = gsm.burst_from_coded(bytes(self.buf)[base:base + BPLEN])
self.mask |= 1 << bid
self.tx.schedule(tn, fn_air, burst, True)
self.stats.tch_ul_bursts += 1scripts/l1ctl_bridge.py
scripts/l1ctl_bridge.py
#!/usr/bin/env python3
"""
l1ctl_bridge.py — Pure sercomm ↔ UDP bridge.
The bridge is a translator/forwarder:
mobile ←L1CTL→ bridge ←sercomm/UART→ firmware ARM (L1 processing)
BTS ←TRXD/TRXC/CLK→ bridge (built-in TRX server)
The firmware ARM does ALL L1 processing. The bridge just forwards.
Bursts from BTS arrive via TRXD → bridge sends to firmware via sercomm (new DLCI).
TX bursts from firmware arrive via sercomm → bridge sends to BTS via TRXD.
"""
import errno, fcntl, os, select, signal, socket, struct, sys, termios, threading, time
# ═══════════════════════════════════════════════════════════════════════════════
# Constants
# ═══════════════════════════════════════════════════════════════════════════════
GSM_HYPERFRAME = 2715648
GSM_FRAME_US = 4615.0
L1CTL_SOCK = "/tmp/osmocom_l2"
SERCOMM_FLAG = 0x7E; SERCOMM_ESCAPE = 0x7D; SERCOMM_XOR = 0x20
DLCI_L1CTL = 5
DLCI_BURST = 6 # New DLCI for burst data (firmware ↔ bridge)
# ═══════════════════════════════════════════════════════════════════════════════
# Helpers
# ═══════════════════════════════════════════════════════════════════════════════
def udp_bind_connect(bind_port, dst_addr, dst_port):
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("127.0.0.1", bind_port))
s.connect((dst_addr, dst_port))
s.setblocking(False); return s
def sercomm_wrap(dlci, payload):
out = bytearray([SERCOMM_FLAG])
for b in bytes([dlci, 0x03]) + payload:
if b in (SERCOMM_FLAG, SERCOMM_ESCAPE):
out.append(SERCOMM_ESCAPE); out.append(b ^ SERCOMM_XOR)
else: out.append(b)
out.append(SERCOMM_FLAG); return bytes(out)
class SercommParser:
def __init__(self): self.buf = bytearray()
def feed(self, data):
self.buf.extend(data); frames = []
while True:
try: s = self.buf.index(SERCOMM_FLAG)
except ValueError: self.buf.clear(); break
if s > 0: del self.buf[:s]
try: e = self.buf.index(SERCOMM_FLAG, 1)
except ValueError: break
raw = bytes(self.buf[1:e]); del self.buf[:e+1]
if not raw: continue
out = bytearray(); i = 0
while i < len(raw):
if raw[i] == SERCOMM_ESCAPE: i += 1; out.append(raw[i] ^ SERCOMM_XOR) if i < len(raw) else None
else: out.append(raw[i])
i += 1
if len(out) >= 2: frames.append((out[0], bytes(out[2:])))
return frames
class LPReader:
def __init__(self): self.buf = bytearray()
def feed(self, data):
self.buf.extend(data); msgs = []
while len(self.buf) >= 2:
ml = struct.unpack(">H", self.buf[:2])[0]
if len(self.buf) < 2 + ml: break
msgs.append(bytes(self.buf[2:2+ml])); del self.buf[:2+ml]
return msgs
# ═══════════════════════════════════════════════════════════════════════════════
# BTS TRX Server — built-in virtual TRX for the BTS
# ═══════════════════════════════════════════════════════════════════════════════
class BTSTrx:
"""Virtual TRX for the BTS. Provides CLK, TRXC, TRXD."""
def __init__(self, bts_base=5700):
self.bts_base = bts_base
self.clk_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.clk_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.clk_sock.bind(("127.0.0.1", bts_base)); self.clk_sock.setblocking(False)
self.trxc_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.trxc_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.trxc_sock.bind(("127.0.0.1", bts_base + 1)); self.trxc_sock.setblocking(False)
self.trxd_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.trxd_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.trxd_sock.bind(("127.0.0.1", bts_base + 2)); self.trxd_sock.setblocking(False)
self.bts_trxd_addr = None
self.clk_fn = 0
self.running = False
self.stats = {"dl": 0, "ul": 0}
self._stop = threading.Event()
self._clk_thread = threading.Thread(target=self._clock_run, daemon=True)
self._clk_thread.start()
print(f"bts-trx: listening on {bts_base}-{bts_base+2}", flush=True)
def _clock_run(self):
tick_ns = int(GSM_FRAME_US * 1000)
t_next = time.monotonic_ns()
while not self._stop.is_set():
t_next += tick_ns
dt = t_next - time.monotonic_ns()
if dt > 0: time.sleep(dt / 1e9)
elif dt < -tick_ns * 10: t_next = time.monotonic_ns()
self.clk_fn = (self.clk_fn + 1) % GSM_HYPERFRAME
if self.clk_fn % 102 == 0:
try: self.clk_sock.sendto(f"IND CLOCK {self.clk_fn}\0".encode(),
("127.0.0.1", self.bts_base + 100))
except: pass
def handle_trxc(self):
try: data, addr = self.trxc_sock.recvfrom(256)
except: return
cmd = data.strip(b'\x00').decode(errors='replace')
parts = cmd.split()
if len(parts) < 2 or parts[0] != "CMD": return
verb = parts[1]
params = " ".join(parts[2:])
if verb != "SETPOWER":
print(f" [TRXC] {cmd}", flush=True)
if verb == "POWERON":
self.running = True; rsp = "RSP POWERON 0"
elif verb == "POWEROFF":
self.running = False; rsp = "RSP POWEROFF 0"
elif verb == "SETFORMAT":
rsp = "RSP SETFORMAT 0"
elif params:
rsp = f"RSP {verb} 0 {params}"
else:
rsp = f"RSP {verb} 0"
self.trxc_sock.sendto(rsp.encode(), addr)
def recv_dl_burst(self):
"""Receive DL burst from BTS. Returns raw TRXD packet or None."""
try: data, addr = self.trxd_sock.recvfrom(512)
except: return None
self.bts_trxd_addr = addr
self.stats["dl"] += 1
return data
def send_ul_burst(self, data):
"""Send UL burst to BTS. Data is raw TRXD packet."""
if not self.bts_trxd_addr: return
self.trxd_sock.sendto(data, self.bts_trxd_addr)
self.stats["ul"] += 1
def sockets(self):
return [self.trxc_sock, self.trxd_sock]
def close(self):
self._stop.set()
for s in [self.clk_sock, self.trxc_sock, self.trxd_sock]:
try: s.close()
except: pass
# ═══════════════════════════════════════════════════════════════════════════════
# Main — pure forwarder
# ═══════════════════════════════════════════════════════════════════════════════
def send_to_mobile(client, payload):
if client:
try: client.sendall(struct.pack(">H", len(payload)) + payload); return True
except: return False
return False
def main():
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <uart-sock> [l1ctl-sock] [bts-base]"); sys.exit(1)
pty_path = sys.argv[1]
global L1CTL_SOCK
if len(sys.argv) > 2 and not sys.argv[2].startswith("-"):
L1CTL_SOCK = sys.argv[2]
nums = [a for a in sys.argv[2:] if a.isdigit()]
bts_base = int(nums[0]) if len(nums) > 0 else 5700
try: os.unlink(L1CTL_SOCK)
except FileNotFoundError: pass
# QEMU UART
import stat as stat_mod
qemu_sock = None
if os.path.exists(pty_path) and stat_mod.S_ISSOCK(os.stat(pty_path).st_mode):
qemu_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
qemu_sock.connect(pty_path); qemu_sock.setblocking(False)
pty_fd = qemu_sock.fileno()
else:
pty_fd = os.open(pty_path, os.O_RDWR | os.O_NOCTTY)
attrs = termios.tcgetattr(pty_fd)
attrs[0] = attrs[1] = attrs[3] = 0
attrs[2] = termios.CS8 | termios.CLOCAL | termios.CREAD
attrs[4] = attrs[5] = termios.B115200
attrs[6][termios.VMIN] = 1; attrs[6][termios.VTIME] = 0
termios.tcsetattr(pty_fd, termios.TCSANOW, attrs)
fl = fcntl.fcntl(pty_fd, fcntl.F_GETFL)
fcntl.fcntl(pty_fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
# L1CTL server
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
srv.bind(L1CTL_SOCK); srv.listen(1); srv.setblocking(False)
# BTS TRX server
trx = BTSTrx(bts_base)
print(f"bridge: uart={pty_path} l1ctl={L1CTL_SOCK} bts={bts_base}", flush=True)
print(f" sercomm DLCI {DLCI_L1CTL}=L1CTL, DLCI {DLCI_BURST}=bursts", flush=True)
parser = SercommParser(); lp = LPReader()
client = None; running = True
stats = {"l1ctl_tx": 0, "l1ctl_rx": 0, "burst_dl": 0, "burst_ul": 0}
def shutdown(s, f): nonlocal running; running = False
signal.signal(signal.SIGINT, shutdown); signal.signal(signal.SIGTERM, shutdown)
try:
while running:
rlist = [srv, pty_fd] + trx.sockets()
if client: rlist.append(client)
try: readable, _, _ = select.select(rlist, [], [], 0.5)
except (OSError, ValueError): break
# ── Accept mobile ──
if srv in readable:
conn, _ = srv.accept(); conn.setblocking(False)
if client:
try: client.close()
except: pass
client = conn; lp = LPReader()
print("mobile connected", flush=True)
# ── L1CTL: mobile → firmware (pure forward via sercomm) ──
if client and client in readable:
try: raw = client.recv(4096)
except BlockingIOError: raw = b""
if not raw:
print("mobile disconnected", flush=True)
try: client.close()
except: pass
client = None
else:
for payload in lp.feed(raw):
stats["l1ctl_tx"] += 1
frame = sercomm_wrap(DLCI_L1CTL, payload)
try: os.write(pty_fd, frame)
except OSError: pass
if stats["l1ctl_tx"] <= 10:
mt = payload[0] if payload else 0
print(f" [L1CTL→fw] type=0x{mt:02x} len={len(payload)}", flush=True)
# ── UART: firmware → mobile/bridge ──
if pty_fd in readable:
try: raw = os.read(pty_fd, 4096)
except OSError as e:
if e.errno in (errno.EAGAIN, errno.EWOULDBLOCK): raw = b""
else: raise
if raw:
for dlci, payload in parser.feed(raw):
if dlci == DLCI_L1CTL and client:
# L1CTL from firmware → mobile
stats["l1ctl_rx"] += 1
send_to_mobile(client, payload)
elif dlci == DLCI_BURST:
# TX burst from firmware → BTS
trx.send_ul_burst(payload)
stats["burst_ul"] += 1
# ── BTS TRXC ──
if trx.trxc_sock in readable:
trx.handle_trxc()
# ── BTS TRXD: DL burst → firmware via sercomm (TS0, 1 per multiframe) ──
if trx.trxd_sock in readable:
data = trx.recv_dl_burst()
if data and len(data) >= 6 and client:
tn = data[0] & 0x07
fn = struct.unpack(">I", data[1:5])[0]
mf = fn % 51
# Forward FCCH(0,10,20,30,40), SCH(1,11,21,31,41), BCCH(2-5), CCCH(6-9)
if tn == 0 and mf <= 9:
frame = sercomm_wrap(DLCI_BURST, data)
try: os.write(pty_fd, frame)
except OSError: pass
stats["burst_dl"] += 1
finally:
if client: client.close()
srv.close(); trx.close()
if qemu_sock: qemu_sock.close()
else: os.close(pty_fd)
try: os.unlink(L1CTL_SOCK)
except: pass
print(f"done l1ctl={stats['l1ctl_tx']}/{stats['l1ctl_rx']} "
f"burst={stats['burst_dl']}/{stats['burst_ul']}", flush=True)
if __name__ == "__main__":
main()scripts/sms-interop-relay.py
scripts/sms-interop-relay.py
#!/usr/bin/env python3
"""
sms-interop-relay.py — Relay SMS inter-opérateur via proto-smsc-proto
Fixes par rapport à la version précédente :
- Retry HLR VTY lookup avec backoff (race condition au démarrage)
- Retry TCP relay send (3 tentatives avec délai)
- Timeout TCP configurable (connect_timeout dans sms-routing.conf)
- Support N opérateurs distants (pas de limite hardcodée)
- Meilleure gestion des erreurs TPDU
- Log structuré avec contexte opérateur
"""
import os
import sys
import json
import time
import socket
import struct
import select
import signal
import logging
import argparse
import threading
import subprocess
from pathlib import Path
from typing import Optional, Dict, Tuple
# ═══════════════════════════════════════════════════════════════════════════════
# Configuration
# ═══════════════════════════════════════════════════════════════════════════════
RELAY_TCP_PORT = 7890
MO_LOG_POLL_INTERVAL = 0.5
HLR_VTY_HOST = "127.0.0.1"
HLR_VTY_PORT = 4258
SENDMT_SOCKET = "/tmp/sendmt_socket"
# Retry config
HLR_LOOKUP_RETRIES = 3
HLR_LOOKUP_DELAY = 2.0
RELAY_SEND_RETRIES = 3
RELAY_SEND_DELAY = 3.0
RELAY_CONNECT_TIMEOUT = 10
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [RELAY-OP%(operator_id)s] %(levelname)s %(message)s',
datefmt='%Y-%m-%dT%H:%M:%S'
)
# ═══════════════════════════════════════════════════════════════════════════════
# TPDU Parser
# ═══════════════════════════════════════════════════════════════════════════════
def decode_bcd_number(hex_bytes: bytes, num_digits: int) -> str:
number = ""
for byte in hex_bytes:
lo = byte & 0x0F
hi = (byte >> 4) & 0x0F
if lo <= 9:
number += str(lo)
if hi <= 9 and len(number) < num_digits:
number += str(hi)
return number[:num_digits]
def parse_sms_submit_tpdu(hex_str: str) -> Optional[Dict]:
try:
data = bytes.fromhex(hex_str.strip())
except ValueError:
return None
if len(data) < 4:
return None
first_octet = data[0]
mti = first_octet & 0x03
if mti != 0x01:
return None
vpf = (first_octet >> 3) & 0x03
mr = data[1]
da_len = data[2]
da_toa = data[3]
da_bytes = (da_len + 1) // 2
if len(data) < 4 + da_bytes:
return None
da_digits = decode_bcd_number(data[4:4 + da_bytes], da_len)
ton = (da_toa >> 4) & 0x07
prefix = "+" if ton == 1 else ""
offset = 4 + da_bytes
result = {
"mti": "SMS-SUBMIT",
"mr": mr,
"da_number": f"{prefix}{da_digits}",
"da_ton": ton,
"da_npi": da_toa & 0x0F,
"da_raw": da_digits,
}
if offset < len(data):
result["tp_pid"] = data[offset]; offset += 1
if offset < len(data):
result["tp_dcs"] = data[offset]; offset += 1
if vpf == 2 and offset < len(data): offset += 1
elif vpf == 3 and offset + 7 <= len(data): offset += 7
elif vpf == 1 and offset + 7 <= len(data): offset += 7
if offset < len(data):
udl = data[offset]; result["tp_udl"] = udl; offset += 1
dcs = result.get("tp_dcs", 0)
if dcs == 0x00 and offset < len(data):
result["user_data_hex"] = data[offset:].hex()
return result
# ═══════════════════════════════════════════════════════════════════════════════
# MO SMS Log Parser
# ═══════════════════════════════════════════════════════════════════════════════
class MOLogEntry:
def __init__(self):
self.timestamp = ""
self.imsi = ""
self.sm_rp_mr = ""
self.sm_rp_da = ""
self.sm_rp_oa = ""
self.sm_rp_oa_number = ""
self.sm_rp_ui_len = 0
self.tpdu_hex = ""
self.parsed_tpdu = None
@property
def destination_number(self) -> str:
if self.parsed_tpdu:
return self.parsed_tpdu.get("da_raw", "")
return ""
@property
def sender_number(self) -> str:
return self.sm_rp_oa_number
def __repr__(self):
return f"MOLogEntry(from={self.sender_number}, to={self.destination_number}, imsi={self.imsi})"
def parse_mo_log_block(lines: list) -> Optional[MOLogEntry]:
entry = MOLogEntry()
for line in lines:
line = line.strip()
if not line: continue
if line.endswith("Rx MO SM"):
entry.timestamp = line.replace(" Rx MO SM", "")
elif line.startswith("IMSI:"):
entry.imsi = line.split(":", 1)[1].strip()
elif line.startswith("SM-RP-MR:"):
entry.sm_rp_mr = line.split(":", 1)[1].strip()
elif line.startswith("SM-RP-DA:"):
entry.sm_rp_da = line.split(":", 1)[1].strip()
elif line.startswith("SM-RP-OA:"):
parts = line.split()
entry.sm_rp_oa = line.split(":", 1)[1].strip()
if parts:
entry.sm_rp_oa_number = parts[-1]
elif line.startswith("SM-RP-UI:"):
try:
entry.sm_rp_ui_len = int(line.split(":")[1].strip().split()[0])
except (ValueError, IndexError):
pass
elif all(c in "0123456789abcdefABCDEF" for c in line) and len(line) >= 4:
entry.tpdu_hex = line
entry.parsed_tpdu = parse_sms_submit_tpdu(line)
if entry.imsi and entry.tpdu_hex:
return entry
return None
# ═══════════════════════════════════════════════════════════════════════════════
# Routing Table — support N opérateurs
# ═══════════════════════════════════════════════════════════════════════════════
class RoutingTable:
def __init__(self, config_path: str):
self.operators: Dict[str, str] = {}
self.routes: list = []
self.local_operator_id = "1"
self.connect_timeout = RELAY_CONNECT_TIMEOUT
self.retry_count = RELAY_SEND_RETRIES
self.retry_delay = RELAY_SEND_DELAY
self._load(config_path)
def _load(self, path: str):
section = None
try:
with open(path, 'r') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'): continue
if line.startswith('[') and line.endswith(']'):
section = line[1:-1].lower(); continue
if '=' not in line: continue
key, val = line.split('=', 1)
key = key.strip()
val = val.split('#')[0].strip()
if section == 'operators':
# « ip » ou « ip:port ». Le port explicite sert au WAN :
# sur un noeud distant, un seul port public par
# opérateur peut atteindre le relais du bon container.
self.operators[key] = val
elif section == 'routes':
# « op_id » ou « op_id strip=N ». strip = nombre de
# chiffres d'INDICATIF a retirer avant de transmettre :
# le HLR du noeud d'arrivee ne connait que le numero nu.
parts = val.split()
op_id = parts[0] if parts else val
strip = 0
for extra in parts[1:]:
if extra.startswith('strip='):
try:
strip = int(extra.split('=', 1)[1])
except ValueError:
strip = 0
self.routes.append((key, op_id, strip))
elif section == 'local':
if key == 'operator_id': self.local_operator_id = val
elif section == 'relay':
if key == 'connect_timeout':
self.connect_timeout = int(val)
elif key == 'retry_count':
self.retry_count = int(val)
elif key == 'retry_delay':
self.retry_delay = int(val)
self.routes.sort(key=lambda x: len(x[0]), reverse=True)
except FileNotFoundError:
logging.warning(f"Routing config not found: {path}, using defaults")
def lookup(self, msisdn: str) -> Optional[str]:
clean = msisdn.lstrip('+')
for prefix, op_id, _strip in self.routes:
if clean.startswith(prefix):
return op_id
return None
def lookup_strip(self, msisdn: str) -> int:
"""Chiffres d'indicatif a retirer pour cette destination (0 = aucun)."""
clean = msisdn.lstrip('+')
for prefix, _op_id, strip in self.routes:
if clean.startswith(prefix):
return strip
return 0
def is_local(self, msisdn: str) -> bool:
op = self.lookup(msisdn)
return op == self.local_operator_id or op is None
def get_operator_ip(self, op_id: str) -> Optional[str]:
"""IP seule (compatibilite : le port se lit avec get_operator_target)."""
target = self.operators.get(op_id)
return target.split(':', 1)[0] if target else None
def get_operator_target(self, op_id: str) -> Optional[Tuple[str, int]]:
"""(ip, port) — port par defaut RELAY_TCP_PORT si non precise."""
target = self.operators.get(op_id)
if not target:
return None
if ':' in target:
host, _, port = target.partition(':')
try:
return host, int(port)
except ValueError:
return host, RELAY_TCP_PORT
return target, RELAY_TCP_PORT
# ═══════════════════════════════════════════════════════════════════════════════
# HLR VTY — MSISDN → IMSI avec retry
# ═══════════════════════════════════════════════════════════════════════════════
def hlr_msisdn_to_imsi(msisdn: str,
hlr_host: str = HLR_VTY_HOST,
hlr_port: int = HLR_VTY_PORT,
retries: int = HLR_LOOKUP_RETRIES,
delay: float = HLR_LOOKUP_DELAY) -> Optional[str]:
"""Lookup MSISDN→IMSI via HLR VTY with retry on failure."""
clean = msisdn.lstrip('+')
for attempt in range(retries):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5)
s.connect((hlr_host, hlr_port))
def recv_until_prompt(timeout=3):
data = b""
deadline = time.time() + timeout
while time.time() < deadline:
try:
chunk = s.recv(4096)
if not chunk: break
data += chunk
decoded = data.decode("ascii", errors="replace")
lines = decoded.strip().splitlines()
if lines:
last = lines[-1].strip()
if last.endswith(">") or last.endswith("#"):
break
except socket.timeout:
break
return data.decode("ascii", errors="replace")
recv_until_prompt()
s.sendall(b"enable\r\n")
recv_until_prompt()
cmd = f"show subscriber msisdn {clean}\r\n"
s.sendall(cmd.encode())
response = recv_until_prompt()
s.close()
for line in response.splitlines():
stripped = line.strip()
if stripped.startswith("IMSI:"):
imsi = stripped.split(":", 1)[1].strip()
if imsi.isdigit():
logging.info(f"HLR match: MSISDN {clean} → IMSI {imsi}")
return imsi
# IMSI not found (not a connection error) — don't retry
logging.warning(f"MSISDN {clean} not found in HLR (attempt {attempt+1})")
if attempt < retries - 1:
time.sleep(delay)
continue
except (ConnectionRefusedError, socket.timeout, OSError) as e:
logging.warning(f"HLR VTY connection failed (attempt {attempt+1}/{retries}): {e}")
try: s.close()
except: pass
if attempt < retries - 1:
time.sleep(delay)
continue
except Exception as e:
logging.error(f"HLR VTY unexpected error: {e}")
try: s.close()
except: pass
if attempt < retries - 1:
time.sleep(delay)
logging.error(f"HLR lookup failed for MSISDN {clean} after {retries} attempts")
return None
# ═══════════════════════════════════════════════════════════════════════════════
# MT SMS Injection (local)
# ═══════════════════════════════════════════════════════════════════════════════
def inject_mt_sms(dest_imsi: str, message_text: str, from_number: str,
sc_address: str, sendmt_socket: str = SENDMT_SOCKET) -> bool:
# Wait for socket if not present yet (race at startup)
for _ in range(5):
if os.path.exists(sendmt_socket):
break
logging.warning(f"sendmt socket {sendmt_socket} not found, waiting 2s...")
time.sleep(2)
if not os.path.exists(sendmt_socket):
logging.error(f"sendmt socket {sendmt_socket} still missing after wait")
return False
try:
# Escape single quotes in message
safe_msg = message_text.replace("'", "'\\''")
cmd = (
f"sms-encode-text '{safe_msg}' "
f"| gen-sms-deliver-pdu {from_number} "
f"| proto-smsc-sendmt {sc_address} {dest_imsi} {sendmt_socket}"
)
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
if result.returncode == 0:
logging.info(f"MT SMS injected: IMSI={dest_imsi}")
return True
else:
logging.error(f"MT SMS injection failed (rc={result.returncode}): {result.stderr}")
return False
except Exception as e:
logging.error(f"MT SMS injection error: {e}")
return False
# ═══════════════════════════════════════════════════════════════════════════════
# TCP Relay Server
# ═══════════════════════════════════════════════════════════════════════════════
class RelayServer(threading.Thread):
def __init__(self, port: int, operator_id: str, sc_address: str):
super().__init__(daemon=True)
self.port = port
self.operator_id = operator_id
self.sc_address = sc_address
self.running = True
def run(self):
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', self.port))
server.listen(5)
server.settimeout(1.0)
logging.info(f"Relay server listening on port {self.port}")
while self.running:
try:
conn, addr = server.accept()
threading.Thread(target=self._handle_client, args=(conn, addr), daemon=True).start()
except socket.timeout:
continue
except Exception as e:
logging.error(f"Server error: {e}")
server.close()
def _handle_client(self, conn: socket.socket, addr: tuple):
try:
conn.settimeout(10)
data = b""
while True:
chunk = conn.recv(4096)
if not chunk: break
data += chunk
if b'\n' in data: break
if not data:
conn.close(); return
msg = json.loads(data.decode('utf-8').strip())
if msg.get("type") == "mt_sms":
self._handle_mt_sms(msg, conn)
else:
response = {"status": "error", "reason": "unknown message type"}
conn.sendall((json.dumps(response) + '\n').encode())
except Exception as e:
logging.error(f"Client handler error from {addr}: {e}")
finally:
conn.close()
def _handle_mt_sms(self, msg: dict, conn: socket.socket):
dest_msisdn = msg.get("dest_msisdn", "")
from_number = msg.get("from_number", "")
message_text = msg.get("message_text", "")
logging.info(f"Incoming interop MT: {from_number} → {dest_msisdn}")
# Resolve with retries
dest_imsi = hlr_msisdn_to_imsi(dest_msisdn)
if not dest_imsi:
response = {"status": "error", "reason": f"MSISDN {dest_msisdn} not found in HLR"}
conn.sendall((json.dumps(response) + '\n').encode())
return
success = inject_mt_sms(
dest_imsi=dest_imsi,
message_text=message_text,
from_number=from_number,
sc_address=self.sc_address
)
response = {
"status": "ok" if success else "error",
"dest_imsi": dest_imsi,
"reason": "" if success else "injection failed"
}
conn.sendall((json.dumps(response) + '\n').encode())
# ═══════════════════════════════════════════════════════════════════════════════
# TCP Relay Client — avec retry
# ═══════════════════════════════════════════════════════════════════════════════
def send_interop_mt(target_ip: str, target_port: int, dest_msisdn: str,
from_number: str, message_text: str, sc_address: str,
sender_imsi: str = "",
retries: int = RELAY_SEND_RETRIES,
delay: float = RELAY_SEND_DELAY,
connect_timeout: int = RELAY_CONNECT_TIMEOUT) -> bool:
"""Send interop MT SMS with retry on TCP failure."""
msg = {
"type": "mt_sms",
"dest_msisdn": dest_msisdn,
"from_number": from_number,
"message_text": message_text,
"sc_address": sc_address,
"sender_imsi": sender_imsi,
}
for attempt in range(retries):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(connect_timeout)
s.connect((target_ip, target_port))
s.sendall((json.dumps(msg) + '\n').encode())
data = b""
while True:
chunk = s.recv(4096)
if not chunk: break
data += chunk
if b'\n' in data: break
s.close()
if data:
response = json.loads(data.decode('utf-8').strip())
if response.get("status") == "ok":
logging.info(f"Interop MT delivered: {from_number} → {dest_msisdn}")
return True
else:
logging.error(f"Interop MT rejected: {response.get('reason')}")
# Don't retry if the remote explicitly rejected
return False
else:
logging.warning(f"Empty response from relay (attempt {attempt+1})")
except (ConnectionRefusedError, socket.timeout, OSError) as e:
logging.warning(f"Relay send to {target_ip}:{target_port} failed "
f"(attempt {attempt+1}/{retries}): {e}")
try: s.close()
except: pass
if attempt < retries - 1:
logging.info(f"Retrying in {delay}s...")
time.sleep(delay)
logging.error(f"Interop send failed after {retries} attempts to {target_ip}")
return False
# ═══════════════════════════════════════════════════════════════════════════════
# GSM 7-bit Decoder
# ═══════════════════════════════════════════════════════════════════════════════
GSM7_BASIC = (
"@£$¥èéùìòÇ\nØø\rÅå"
"Δ_ΦΓΛΩΠΨΣΘΞ\x1bÆæßÉ"
" !\"#¤%&'()*+,-./"
"0123456789:;<=>?"
"¡ABCDEFGHIJKLMNO"
"PQRSTUVWXYZÄÖÑܧ"
"¿abcdefghijklmno"
"pqrstuvwxyzäöñüà"
)
def decode_gsm7(data: bytes, num_chars: int) -> str:
chars = []
for i in range(num_chars):
byte_offset = (i * 7) // 8
bit_offset = (i * 7) % 8
if byte_offset >= len(data): break
val = (data[byte_offset] >> bit_offset) & 0x7F
if bit_offset > 1 and byte_offset + 1 < len(data):
val = ((data[byte_offset] >> bit_offset) |
(data[byte_offset + 1] << (8 - bit_offset))) & 0x7F
if val < len(GSM7_BASIC):
chars.append(GSM7_BASIC[val])
else:
chars.append('?')
return ''.join(chars)
def extract_text_from_tpdu(tpdu_hex: str) -> str:
parsed = parse_sms_submit_tpdu(tpdu_hex)
if not parsed: return ""
dcs = parsed.get("tp_dcs", 0)
udl = parsed.get("tp_udl", 0)
ud_hex = parsed.get("user_data_hex", "")
if dcs == 0x00 and ud_hex:
try:
return decode_gsm7(bytes.fromhex(ud_hex), udl)
except: pass
return ""
def extract_text_via_sms_decode(tpdu_hex: str) -> str:
try:
result = subprocess.run(
['sms-pdu-decode', '-n'],
input=tpdu_hex + '\n',
capture_output=True, text=True, timeout=5
)
if result.returncode == 0:
lines = result.stdout.strip().split('\n')
text_lines = []
found_length = False
for line in lines:
if found_length and line.strip():
text_lines.append(line)
if line.strip().startswith('Length:'):
found_length = True
if text_lines:
return '\n'.join(text_lines).strip()
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
return extract_text_from_tpdu(tpdu_hex)
# ═══════════════════════════════════════════════════════════════════════════════
# MO Log Watcher
# ═══════════════════════════════════════════════════════════════════════════════
class MOLogWatcher(threading.Thread):
def __init__(self, log_path: str, routing: RoutingTable,
operator_id: str, sc_address: str):
super().__init__(daemon=True)
self.log_path = log_path
self.routing = routing
self.operator_id = operator_id
self.sc_address = sc_address
self.running = True
self._position = 0
def run(self):
logging.info(f"Watching MO log: {self.log_path}")
while self.running and not Path(self.log_path).exists():
time.sleep(2)
if not self.running: return
with open(self.log_path, 'r') as f:
f.seek(0, 2)
self._position = f.tell()
logging.info(f"MO log watcher started (pos={self._position})")
while self.running:
try:
self._check_new_entries()
except Exception as e:
logging.error(f"MO watcher error: {e}")
time.sleep(MO_LOG_POLL_INTERVAL)
def _check_new_entries(self):
try:
with open(self.log_path, 'r') as f:
f.seek(self._position)
new_data = f.read()
self._position = f.tell()
except FileNotFoundError:
return
if not new_data: return
blocks = []
current_block = []
for line in new_data.split('\n'):
if "Rx MO SM" in line and current_block:
blocks.append(current_block)
current_block = [line]
else:
current_block.append(line)
if current_block:
blocks.append(current_block)
for block in blocks:
entry = parse_mo_log_block(block)
if entry:
self._route_mo_sms(entry)
def _route_mo_sms(self, entry: MOLogEntry):
dest = entry.destination_number
sender = entry.sender_number
if not dest:
logging.warning(f"MO SMS without parsable destination: {entry}")
return
logging.info(f"MO SMS: {sender} → {dest} (from IMSI {entry.imsi})")
target_op = self.routing.lookup(dest)
if target_op is None:
logging.warning(f"No route for destination {dest}")
return
if target_op == self.operator_id:
logging.info(f"Local delivery for {dest}")
self._deliver_local(entry, dest)
return
target = self.routing.get_operator_target(target_op)
if not target:
logging.error(f"No IP for operator {target_op}")
return
target_ip, target_port = target
# Indicatif WAN : on le retire ICI, cote emetteur. Le noeud d'arrivee
# recoit le numero nu, celui que son HLR connait ; sans ce retrait il
# cherche « 3310001 » et repond « MSISDN not found in HLR ».
strip = self.routing.lookup_strip(dest)
dest_out = dest.lstrip('+')[strip:] if strip else dest
if strip:
logging.info(f"WAN strip {strip}: {dest} → {dest_out}")
logging.info(f"Interop route: {dest} → Op{target_op} ({target_ip}:{target_port})")
message_text = extract_text_via_sms_decode(entry.tpdu_hex)
if not message_text:
message_text = extract_text_from_tpdu(entry.tpdu_hex)
if not message_text:
logging.warning(f"Could not decode text from TPDU: {entry.tpdu_hex[:40]}...")
message_text = "[undecoded]"
success = send_interop_mt(
target_ip=target_ip,
target_port=target_port,
dest_msisdn=dest_out,
from_number=sender,
message_text=message_text,
sc_address=self.sc_address,
sender_imsi=entry.imsi,
retries=self.routing.retry_count,
delay=self.routing.retry_delay,
connect_timeout=self.routing.connect_timeout,
)
if success:
logging.info(f"✓ Interop SMS: {sender} → {dest} (Op{target_op})")
else:
logging.error(f"✗ Interop SMS failed: {sender} → {dest} (Op{target_op})")
def _deliver_local(self, entry: MOLogEntry, dest_msisdn: str):
dest_imsi = hlr_msisdn_to_imsi(dest_msisdn)
if not dest_imsi:
logging.warning(f"Local MSISDN {dest_msisdn} not found in HLR")
return
message_text = extract_text_via_sms_decode(entry.tpdu_hex)
if not message_text:
message_text = extract_text_from_tpdu(entry.tpdu_hex)
if not message_text:
message_text = "[undecoded]"
inject_mt_sms(
dest_imsi=dest_imsi,
message_text=message_text,
from_number=entry.sender_number,
sc_address=self.sc_address
)
# ═══════════════════════════════════════════════════════════════════════════════
# Main
# ═══════════════════════════════════════════════════════════════════════════════
def main():
parser = argparse.ArgumentParser(description='SMS Interop Relay')
parser.add_argument('--config', '-c', default='/etc/osmocom/sms-routing.conf')
parser.add_argument('--port', '-p', type=int, default=RELAY_TCP_PORT)
parser.add_argument('--mo-log')
parser.add_argument('--operator-id', default=os.environ.get('OPERATOR_ID', '1'))
# --node-id : LA CLE DE ROUTAGE, et elle n'est pas l'operateur.
#
# local_operator_id etait ecrase par --operator-id juste apres la lecture du
# fichier, donc la cle du relais valait toujours le numero d'OPERATEUR. Avec
# des MSISDN qui commencent par le noeud (<noeud>00<op><ms>), sms-routing.conf
# est desormais indexe par NOEUD : l'operateur 1 du noeud 1 et celui du noeud
# 2 portaient sinon la meme cle "1", et chacun tenait les numeros de l'autre
# pour locaux - un SMS interop n'etait jamais emis, il etait injecte dans le
# HLR local qui ne connait pas l'abonne.
#
# L'operateur, lui, garde ce qui le concerne : la SC-address (le numero du
# centre SMS, qu'un abonne voit) et le chemin du journal MO. Les changer
# renommerait un fichier que d'autres outils suivent, et un numero que des
# mobiles ont deja en memoire.
parser.add_argument('--node-id',
default=os.environ.get('OSMO_WAN_NODE',
os.environ.get('WAN_NODE_ID', '')))
args = parser.parse_args()
op_id = args.operator_id
node_id = args.node_id if args.node_id not in ('', '0') else op_id
sc_address = f"1999001{op_id}444"
mo_log = args.mo_log or f"/var/log/osmocom/mo-sms-op{op_id}.log"
old_factory = logging.getLogRecordFactory()
def record_factory(*a, **kw):
record = old_factory(*a, **kw)
record.operator_id = op_id
return record
logging.setLogRecordFactory(record_factory)
logging.info(f"SMS Interop Relay starting")
logging.info(f" Operator : {op_id}")
logging.info(f" Node : {node_id} (cle de routage dans sms-routing.conf)")
logging.info(f" SC-address : {sc_address}")
logging.info(f" MO log : {mo_log}")
logging.info(f" TCP port : {args.port}")
routing = RoutingTable(args.config)
routing.local_operator_id = node_id
logging.info(f" Operators : {routing.operators}")
logging.info(f" Routes : {len(routing.routes)} entries")
logging.info(f" Retry : {routing.retry_count}x, delay={routing.retry_delay}s, timeout={routing.connect_timeout}s")
server = RelayServer(port=args.port, operator_id=op_id, sc_address=sc_address)
server.start()
watcher = MOLogWatcher(log_path=mo_log, routing=routing, operator_id=op_id, sc_address=sc_address)
watcher.start()
def signal_handler(sig, frame):
logging.info("Shutting down...")
watcher.running = False
server.running = False
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
logging.info("Relay running. Ctrl+C to stop.")
while True:
time.sleep(1)
if __name__ == '__main__':
main()ss7-console.py
tools/grgsm_relay_decode.py
tools/grgsm_relay_decode.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# tools/grgsm_relay_decode.py — décodeur gr-gsm sur l'I/Q CONTINU relayée (mode
# full-grgsm). Lit l'I/Q du BTS relayée par calypso-ipc-device (UDP 5810,
# fc32), la passe dans le receiver gr-gsm complet (gsm.receiver : FCCH/SCH
# sync + démod GMSK), décode le BCCH/CCCH, et envoie le L2 en GSMTAP au
# listener du shunt (4730) → feed_si → a_cd → le mobile (sur osmocon) campe.
#
# C'est un DÉCODEUR (pas un transceiver) : pas besoin de trxcon. La chaîne
# est celle de grgsm_decode mais avec une source UDP au lieu d'un fichier.
#
# udp_source(5810, fc32) → lpf → gsm.receiver(osr) → C0 bursts →
# gsm_bcch_ccch_demapper → control_channels_decoder → GSMTAP(4730)
#
# osr DOIT matcher le SPS d'osmo-trx (tx-sps=1 → osr=1). Si la sync n'accroche
# pas à osr=1, passer osmo-trx en SPS=4 et CALYPSO_TRX_OSR=4.
import os, sys
from gnuradio import gr, network, filter
from gnuradio.filter import firdes
from gnuradio.fft import window
from gnuradio import gsm
# OSR = oversampling vu par gsm.receiver. Il a besoin de ~4 pour locker la
# FCCH/SCH et la récupération de timing GMSK. L'I/Q relayée est à 1 SPS
# (osmo-trx tx-sps=1) → on RESAMPLE 1→OSR dans le flowgraph (pas de chirurgie
# device, pas de SPS=4 côté osmo-trx).
OSR = int(os.environ.get("CALYPSO_TRX_OSR", "4"))
IN_SPS = int(os.environ.get("CALYPSO_TRX_IN_SPS", "1")) # SPS du relais (osmo-trx)
SYM_RATE = 1625000.0 / 6.0 # 270833.33 Hz
SAMP_RATE = SYM_RATE * OSR # rate après resampling
RX_PORT = int(os.environ.get("CALYPSO_TRX_IQ_RX_PORT", "5810"))
GSMTAP_HOST = os.environ.get("GSMTAP_HOST", "127.0.0.1")
GSMTAP_PORT = int(os.environ.get("CALYPSO_SHUNT_GSMTAP_PORT", "4730"))
TS = int(os.environ.get("TS", "0"))
tb = gr.top_block("grgsm relay decode")
# Source : I/Q continu relayé (fc32, 625 samples/datagram = 5000 o). src_zeros
# → flux continu si pas de data (le receiver ne stalle pas).
src = network.udp_source(gr.sizeof_gr_complex, 1, RX_PORT,
0, 5000, False, True, False)
# Resampling 1→OSR (IN_SPS→OSR) : donne à gsm.receiver les ~4 SPS qu'il attend.
resamp = filter.rational_resampler_ccc(interpolation=OSR, decimation=IN_SPS)
lpf = filter.fir_filter_ccf(1, firdes.low_pass(
1, SAMP_RATE, 125e3, 5e3, window.WIN_HAMMING, 6.76))
rx = gsm.receiver(OSR, [0], []) # cell_allocation [0], tseq []
dm = gsm.gsm_bcch_ccch_demapper(TS)
dec = gsm.control_channels_decoder()
snk = network.socket_pdu("UDP_CLIENT", GSMTAP_HOST, str(GSMTAP_PORT), 10000)
tb.connect((src, 0), (resamp, 0))
tb.connect((resamp, 0), (lpf, 0))
tb.connect((lpf, 0), (rx, 0))
tb.msg_connect((rx, 'C0'), (dm, 'bursts'))
tb.msg_connect((dm, 'bursts'), (dec, 'bursts'))
tb.msg_connect((dec, 'msgs'), (snk, 'pdus'))
print("[grgsm-relay-decode] I/Q udp:%d (osr=%d, %.0f Hz) -> GSMTAP %s:%d -> feed_si"
% (RX_PORT, OSR, SAMP_RATE, GSMTAP_HOST, GSMTAP_PORT), flush=True)
tb.start()
try:
tb.wait()
except KeyboardInterrupt:
tb.stop(); tb.wait()tools/qemu_bcch_grgsm.py
tools/qemu_bcch_grgsm.py
#!/usr/bin/env python3
"""
tools/qemu_bcch_grgsm.py — pont gr-gsm : I/Q DL du BTS (tee qemu) → bursts → demapper
+ decoder gr-gsm → GSMTAP → le shunt qemu (4730) → a_cd → le mobile campe.
ZÉRO hack : le VRAI signal du BTS, démodulé et décodé par la chaîne gr-gsm
officielle (gsm_bcch_ccch_demapper + control_channels_decoder, le même code
que grgsm_decode).
Chaîne :
UDP I/Q (tee BSP qemu, port 6703) ──parse TRXD (8 hdr + 148 cs16)──>
──GMSK demod différentiel (1 SPS)──> 148 bits durs
──gsmtap_hdr(UM_BURST) + 148 octets──> post au demapper gr-gsm
──gsm_bcch_ccch_demapper (assemble 4 bursts du 51-MF)──>
──control_channels_decoder (deinterleave+Viterbi+FIRE)──> GSMTAP L2
──network.socket_pdu(UDP_CLIENT 127.0.0.1:4730)──> shunt feed_si → a_cd
Le démod produit les 148 bits du burst complet (3 tail | 57 | 1 | 26 train |
1 | 57 | 3 tail) ; le decoder gr-gsm extrait/corrige tout seul. Si décode jamais
(le decoder est silencieux sur CRC fail), inverser BIT_SIGN.
Usage : source /root/.env/bin/activate ; python tools/qemu_bcch_grgsm.py
Env : IQ_TEE_PORT(6703) GSMTAP_HOST/PORT(127.0.0.1/4730) ARFCN(514)
TS(0) BIT_SIGN(1)
"""
import os, socket, struct
import numpy as np
from gnuradio import gr, gsm, network
IQ_TEE_PORT = int(os.environ.get("IQ_TEE_PORT", "6703"))
GSMTAP_HOST = os.environ.get("GSMTAP_HOST", "127.0.0.1")
GSMTAP_PORT = int(os.environ.get("GSMTAP_PORT", "4730"))
ARFCN = int(os.environ.get("ARFCN", "514"))
TS = int(os.environ.get("TS", "0"))
BIT_SIGN = int(os.environ.get("BIT_SIGN", "1"))
# --- gsmtap.h (gr-gsm) ---
GSMTAP_VERSION = 0x02
GSMTAP_TYPE_UM_BURST = 0x03
GSMTAP_BURST_NORMAL = 0x06
BURST_SIZE = 148
# Les 8 training sequences GSM (26 bits) — pour valider le démod live.
ALL_TSC = [
"00100101110000100010010111", "00101101110111100010110111",
"01000011101110100100001110", "01000111101101000100011110",
"00011010111001000001101011", "01001110101100000100111010",
"10100111110110001010011111", "11101111000100101110111100",
]
dbg_tsc = [0] # compteur de bursts BCCH instrumentés (mutable)
# [2026-08-04] Manquait : le bloc de dump I/Q plus bas (udp_loop) faisait
# dbg_dump.append(...) sans que la liste existe -> NameError a la PREMIERE
# rafale BCCH (fn%51 dans 2..5), donc pont demarre puis mort aussitot.
dbg_dump = [] # bursts BCCH bruts en attente de vidage (mutable)
def demod_burst(iq):
"""iq : np.complex (>=148) -> string de 148 bits durs '0'/'1', ou None.
Démod GMSK validé bout-en-bout sur capture réelle (26/26 TSC, FIRE CRC ok) :
- dérotation -pi/2 par sample (y = s * exp(-j pi/2 k))
- slicer sur imag(y)>0
- inversion de polarité (BIT_SIGN=1 -> invert ; -1 -> non)
- décalage gauche d'1 bit : le burst osmo-trx arrive avec TSC@62,
gr-gsm attend TSC@61 (3 tail|57|1|26 TSC@61..86|1|57|3 tail).
"""
if len(iq) < BURST_SIZE:
return None
s = iq[:BURST_SIZE].astype(np.complex64)
k = np.arange(BURST_SIZE)
y = s * np.exp(-1j * np.pi / 2.0 * k) # dérotation -pi/2
bits = (y.imag > 0).astype(int)
if BIT_SIGN > 0:
bits = 1 - bits # polarité (capture réelle: inversée)
bits = np.concatenate([bits[1:], [bits[-1]]]) # shift -1 -> TSC@61
return "".join(str(b) for b in bits)
def decode_block(fns, datas):
"""Run one-shot : burst_source(4 bursts BCCH) -> demapper -> decoder ->
socket_pdu(GSMTAP 4730). gr-gsm officiel, pas de _post. fns/datas = listes
parallèles (frame numbers + strings de 148 bits)."""
tb = gr.top_block("bcch block")
src = gsm.burst_source(fns, [TS] * len(fns), datas)
src.set_arfcn(ARFCN)
dm = gsm.gsm_bcch_ccch_demapper(TS)
dec = gsm.control_channels_decoder()
snk = network.socket_pdu("UDP_CLIENT", GSMTAP_HOST, str(GSMTAP_PORT), 10000)
tb.msg_connect((src, "out"), (dm, "bursts"))
tb.msg_connect((dm, "bursts"), (dec, "bursts"))
tb.msg_connect((dec, "msgs"), (snk, "pdus"))
tb.start()
tb.wait() # burst_source émet puis poste 'done'
def udp_loop():
rx = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
rx.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
rx.bind(("0.0.0.0", IQ_TEE_PORT))
print(f"[grgsm] I/Q in udp:{IQ_TEE_PORT} -> GSMTAP {GSMTAP_HOST}:{GSMTAP_PORT} "
f"arfcn={ARFCN} ts={TS}", flush=True)
# On NE filtre PAS par mf : on nourrit le demapper avec un 51-multiframe
# COMPLET (toutes positions). Le demapper gr-gsm mappe lui-même BCCH/CCCH
# selon fn%51 et extrait SI1/2/3/4 à la vraie position planifiée (TC).
# (Le démod produit du bruit aux positions FCCH/SCH, que le demapper ignore.)
# On décode par FENÊTRE de WINDOW_MF multiframes (pas 1 seul) : le demapper
# gr-gsm a besoin d'un run CONTIGU de plusieurs 51-multiframes pour locker
# son compteur TC et extraire les blocs BCCH. 1 multiframe isolé → 0 décode
# (régression). 12 mf couvrent un cycle TC complet (0..7) → SI1/2/3/4 inclus.
WINDOW_MF = int(os.environ.get("WINDOW_MF", "12"))
cur_mf = None # index multiframe courant (fn//51)
fns, datas = [], []
mf_count = 0
npkt = nfed = ndec = 0
while True:
pkt, _ = rx.recvfrom(4096)
npkt += 1
if len(pkt) < 8 + BURST_SIZE * 4:
if npkt <= 3:
print(f"[grgsm] paquet court len={len(pkt)} (attendu>={8+BURST_SIZE*4})", flush=True)
continue
fn = struct.unpack(">I", pkt[1:5])[0]
raw = np.frombuffer(pkt[8:8 + BURST_SIZE * 4], dtype="<i2").astype(np.float32)
bits = demod_burst(raw[0::2] + 1j * raw[1::2])
if bits is None:
continue
nfed += 1
if nfed <= 5 or nfed % 1000 == 0:
print(f"[grgsm] burst #{nfed} fn={fn} mf={fn%51} pwr={np.mean(np.abs(raw)):.0f}", flush=True)
# DEBUG : valide le démod sur les bursts BCCH live (TSC7 @61). Si le
# match n'est pas ~26/26, le démod est faux sur le signal courant →
# explique 0 GSMTAP. Compare aussi les 8 TSC pour repérer un décalage.
if (fn % 51) in (2, 3, 4, 5) and dbg_tsc[0] < 24:
dbg_tsc[0] += 1
mid = bits[61:87]
best = max((sum(a == b for a, b in zip(mid, t)), i)
for i, t in enumerate(ALL_TSC))
print(f"[grgsm][TSC] fn={fn} mf={fn%51} best TSC{best[1]}={best[0]}/26 "
f"(TSC7={sum(a==b for a,b in zip(mid, ALL_TSC[7]))}/26) mid={mid}", flush=True)
# DUMP I/Q brut des bursts BCCH pour brute-force offline du démod.
dbg_dump.append((int(fn), raw.copy()))
if len(dbg_dump) >= 8:
import numpy as _np
d = {}
for i, (f, r) in enumerate(dbg_dump):
d[f"b{i}_iq"] = r.astype(_np.int16); d[f"b{i}_fn"] = f
_np.savez("/opt/GSM/live_bursts.npz", **d)
print(f"[grgsm][DUMP] 8 bursts BCCH live -> /opt/GSM/live_bursts.npz", flush=True)
dbg_dump.clear()
mfi = fn // 51
if cur_mf is None:
cur_mf = mfi
if mfi != cur_mf: # nouvelle frontière de multiframe
mf_count += 1
cur_mf = mfi
if mf_count >= WINDOW_MF and fns: # assez de contexte -> décode le lot
ndec += 1
try:
decode_block(fns, datas) # demapper -> GSMTAP (SI) si CRC ok
if ndec <= 5 or ndec % 20 == 0:
print(f"[grgsm] décode #{ndec} ({len(fns)} bursts / {mf_count} mf, "
f"fn {fns[0]}..{fns[-1]})", flush=True)
except Exception as e:
print(f"[grgsm] decode_block err: {e}", flush=True)
fns, datas = [], []
mf_count = 0
fns.append(fn)
datas.append(bits)
def main():
udp_loop()
if __name__ == "__main__":
main()tools/si.py
tools/si.py
#!/usr/bin/env python3
# Bridge : grgsm_decode -v (qui DECODE le vrai SI du cfile continu) -> parse les
# lignes SI (PD=0x06) -> GSMTAP 16B + L2 -> 4730 (shunt feed_si -> DATA_IND -> mobile).
import subprocess, socket, struct, re, sys
GSMTAP = struct.pack(">BBBBHbbIBBBB", 2,4,0x01,0,0,0,0,0,0x01,0,0,0) # 16B, type UM, BCCH
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
CF = sys.argv[1] if len(sys.argv)>1 else "/tmp/iq_grgsm.fifo"
p = subprocess.Popen(["grgsm_decode","-m","BCCH","-t","0","-a","514","-c",CF,"-s","1083333","-v"],
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
SI={0x1b:"SI3",0x1a:"SI2",0x1c:"SI4",0x21:"SI13",0x19:"SI1",0x1d:"SI2bis",0x1e:"SI2ter",0x06:"RR"}
n=0; nsch=0
for line in p.stdout:
# SCH REEL : le receiver gr-gsm patche imprime "SCHBSIC <bsic> <fn> <toa>" sur
# stdout (le port measurements est avale par le CLI grgsm_decode). On le
# forwarde au shunt en UDP 4731 {magic 'SCH1', int32 bsic, int32 fn, int32 toa}
# -> feed_sb -> shunt_dispatch_sb (BSIC reel; TOA reel si hors CALYPSO_CANNED).
msch = re.search(r"SCHBSIC\s+(\d+)\s+(\d+)\s+(-?\d+)", line)
if msch:
bsic, fn, toa = int(msch.group(1)), int(msch.group(2)), int(msch.group(3))
s.sendto(b"SCH1" + struct.pack("<iii", bsic, fn, toa), ("127.0.0.1", 4731))
nsch += 1
if nsch<=10 or nsch%100==0:
print("[si-bridge] SCH -> shunt 4731 : BSIC=%d (ncc=%d bcc=%d) FN=%d TOA=%d #%d"
%(bsic,(bsic>>3)&7,bsic&7,fn,toa,nsch),flush=True)
continue
m = re.search(r":\s+([0-9a-fA-F][0-9a-fA-F ]+)$", line.strip())
if not m: continue
try: by = bytes(int(x,16) for x in m.group(1).split())
except: continue
if len(by) >= 3 and by[1]==0x06 and by[2] in SI: # RR PD + SI type
L2 = (by[:23] + b"\x2b"*23)[:23]
s.sendto(GSMTAP + L2, ("127.0.0.1",4730))
n+=1
if n<=20 or n%50==0:
print("[si-bridge] %s (mt=0x%02x) -> feed_si (4730) #%d"%(SI[by[2]],by[2],n),flush=True)
print("[si-bridge] fini, %d SI transmis"%n,flush=True)