1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
|
/* PCU Interface codec poart in TTCN-3
* (C) 2018 Harald Welte <laforge@gnumonks.org>
* contributions by sysmocom - s.f.m.c. GmbH
* All rights reserved.
*
* Released under the terms of GNU General Public License, Version 2 or
* (at your option) any later version.
*
* SPDX-License-Identifier: GPL-2.0-or-later
*/
module PCUIF_CodecPort {
import from Osmocom_Types all;
import from PCUIF_Types all;
import from UD_PortType all;
import from UD_Types all;
import from General_Types all;
type record PCUIF_send_data {
PCUIF_Message data,
integer id
};
private function PCUIF_to_UD(in PCUIF_send_data pin, out UD_send_data pout) {
pout.id := pin.id;
pout.data := enc_PCUIF_Message(pin.data);
} with { extension "prototype(fast)" };
private function fix_padding(inout PCUIF_data data) {
data.data := substr(data.data, 0, data.len);
}
private function UD_to_PCUIF(in UD_send_data pin, out PCUIF_send_data pout) {
pout.id := pin.id;
pout.data := dec_PCUIF_Message(pin.data);
/* HACK: fix padding in decoded message. Due to a bug in TITAN, we
* cannot just use its 'PADDING' attribute because it breaks decoding. */
if (ischosen(pout.data.u.data_req)) { fix_padding(pout.data.u.data_req); }
if (ischosen(pout.data.u.data_cnf)) { fix_padding(pout.data.u.data_cnf); }
if (ischosen(pout.data.u.data_ind)) { fix_padding(pout.data.u.data_ind); }
} with { extension "prototype(fast)" };
type port PCUIF_CODEC_PT message {
out UD_close, UD_listen, UD_shutdown, UD_connect, PCUIF_send_data;
in UD_listen_result, UD_connect_result, UD_connected, PCUIF_send_data;
} with { extension "user UD_PT
out (
UD_close -> UD_close:simple;
UD_listen -> UD_listen:simple;
UD_shutdown -> UD_shutdown:simple;
UD_connect -> UD_connect:simple;
PCUIF_send_data -> UD_send_data: function(PCUIF_to_UD)
)
in (
UD_listen_result -> UD_listen_result:simple;
UD_connect_result -> UD_connect_result:simple;
UD_send_data -> PCUIF_send_data: function(UD_to_PCUIF);
UD_connected -> UD_connected:simple
)"
};
template PCUIF_send_data t_SD_PCUIF(integer id, template PCUIF_Message pdu) := {
data := pdu,
id := id
}
template PCUIF_send_data t_SD_PCUIF_MSGT(integer id, template PCUIF_MsgType msg_type,
template uint8_t bts_nr := ?) := {
data := {
msg_type := msg_type,
bts_nr := bts_nr,
spare := ?,
u := ?
},
id := id
}
function f_pcuif_connect(PCUIF_CODEC_PT pt, charstring sock) return integer {
var UD_connect_result res;
timer T := 5.0;
T.start;
pt.send(UD_connect:{sock, -1});
alt {
[] pt.receive(UD_connect_result:?) -> value res {
if (ispresent(res.result) and ispresent(res.result.result_code) and
res.result.result_code == ERROR) {
if (ispresent(res.result.err)) {
setverdict(fail, "Error connecting to PCU socket ", sock, ": ", res.result.err);
} else {
setverdict(fail, "Error connecting to PCU socket ", sock);
}
mtc.stop;
} else {
return res.id;
}
}
[] T.timeout {
setverdict(fail, "Timeout connecting to PCU socket ", sock);
mtc.stop;
}
}
return -23;
}
function f_pcuif_close(PCUIF_CODEC_PT pt, integer id)
{
pt.send(UD_close:{id := id});
}
function f_pcuif_listen(PCUIF_CODEC_PT pt, charstring sock) return integer {
var UD_listen_result res;
var UD_connected udc;
timer T := 5.0;
pt.send(UD_listen:{sock});
T.start;
alt {
[] pt.receive(UD_listen_result:?) -> value res {
if (ispresent(res.result) and ispresent (res.result.result_code) and
res.result.result_code == ERROR) {
if (ispresent(res.result.err)) {
setverdict(fail, "Error listening on PCU socket ", sock, ": ", res.result.err);
} else {
setverdict(fail, "Error listening on PCU socket ", sock);
}
mtc.stop;
} else {
return res.id;
}
}
[] T.timeout {
setverdict(fail, "Timeout waiting for PCU socket ", sock, " connection");
mtc.stop;
}
}
return -23;
}
function f_PCUIF_tx_imm_ass_pch(PCUIF_CODEC_PT pt, integer conn_id, octetstring imm_ass, hexstring imsi,
uint8_t bts_nr := 0, boolean wait_for_cnf := true, OCT4 msg_id := '01020304'O) {
var PCUIF_send_data sd;
timer T := 3.0;
if (mp_pcuif_version < 11) {
/* append 3 last imsi digits so that the BTS is able to compute paging group */
var hexstring last_digits := substr(imsi, lengthof(imsi)-3, 3);
log("3 last imsi digits: ", last_digits);
var octetstring prefix := ''O;
for (var integer i := 0; i < 3; i := i+1) {
prefix := prefix & int2oct(hex2int('30'H) + hex2int(last_digits[i]), 1);
}
pt.send(t_SD_PCUIF(conn_id,
ts_PCUIF_DATA_REQ(bts_nr, 0, 0, 0, 0, PCU_IF_SAPI_PCH, prefix & imm_ass)));
} else {
var PCUIF_pch pch;
pch.msg_id := msg_id;
pch.imsi := hex2str(imsi);
pch.data := imm_ass;
pt.send(t_SD_PCUIF(conn_id, ts_PCUIF_DATA_REQ(bts_nr, 0, 0, 0, 0, PCU_IF_SAPI_PCH_2, enc_PCUIF_pch(pch))));
}
/* Exit early when the caller is not interested in the confirmation message */
if (wait_for_cnf == false) {
return;
}
T.start;
alt {
[] pt.receive(t_SD_PCUIF(conn_id, tr_PCUIF_DATA_CNF(bts_nr, 0, 0, PCU_IF_SAPI_PCH))) -> value sd {
if (mp_pcuif_version >= 11) {
setverdict(fail, "expecting tr_PCUIF_DATA_CNF_2 in PCUIF v.11 or later");
mtc.stop;
} else {
log("IMM.ASS was sent on PCH at fn ", sd.data.u.data_cnf.fn);
return;
}
}
[] pt.receive(t_SD_PCUIF(conn_id, tr_PCUIF_DATA_CNF_2(bts_nr, PCU_IF_SAPI_PCH_2))) -> value sd {
if (mp_pcuif_version < 11) {
setverdict(fail, "expecting tr_PCUIF_DATA_CNF in PCUIF v.10 or earlier");
mtc.stop;
} else {
log("IMM.ASS was sent on PCH");
return;
}
}
[] pt.receive { repeat; }
[] T.timeout {
setverdict(fail, "Timeout waiting for PCU DATA.cnf (PCH)");
mtc.stop;
}
}
}
}
|
