Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 15 additions & 11 deletions gammu/src/convertors/sms.c
Original file line number Diff line number Diff line change
Expand Up @@ -664,8 +664,9 @@ int UDHFromPython(PyObject * dict, GSM_UDHHeader * udh)
udh->Length = len;

if (udh->Length > GSM_MAX_UDH_LENGTH) {
pyg_warning("UDH too large, truncating!");
udh->Length = GSM_MAX_UDH_LENGTH;
PyErr_Format(PyExc_ValueError, "UDH is too large (maximum is %d bytes)",
GSM_MAX_UDH_LENGTH);
return 0;
}

memcpy(udh->Text, s, udh->Length);
Expand Down Expand Up @@ -790,8 +791,10 @@ int SMSFromPython(PyObject * dict, GSM_SMSMessage * sms, int needslocation,
sms->Length = len;

if (sms->Length > GSM_MAX_SMS_LENGTH) {
pyg_warning("SMS text too large, truncating!\n");
sms->Length = GSM_MAX_SMS_LENGTH;
PyErr_Format(PyExc_ValueError,
"SMS text is too large (maximum is %d bytes)",
GSM_MAX_SMS_LENGTH);
return 0;
}

memcpy(sms->Text, s, sms->Length);
Expand Down Expand Up @@ -1122,9 +1125,10 @@ int MultiSMSFromPython(PyObject * list, GSM_MultiSMSMessage * sms)
len = PyList_Size(list);

if (len > GSM_MAX_MULTI_SMS) {
pyg_warning("Truncating MultiSMS entries to %d entries! (from %"
PY_FORMAT_SIZE_T "d))\n", GSM_MAX_MULTI_SMS, len);
len = GSM_MAX_MULTI_SMS;
PyErr_Format(PyExc_ValueError,
"MultiSMS has too many entries (maximum is %d)",
GSM_MAX_MULTI_SMS);
return 0;
}
sms->Number = len;

Expand Down Expand Up @@ -1993,10 +1997,10 @@ int SMSInfoFromPython(PyObject * dict, GSM_MultiPartSMSInfo * entry)
len = PyList_Size(o);

if (len > GSM_MAX_MULTI_SMS - 1) {
pyg_warning("Too many entries, truncating from %"
PY_FORMAT_SIZE_T "d to %d\n", len,
GSM_MAX_MULTI_SMS - 1);
len = GSM_MAX_MULTI_SMS - 1;
PyErr_Format(PyExc_ValueError,
"Too many SMS info entries (maximum is %d)",
GSM_MAX_MULTI_SMS - 1);
return 0;
}

entry->EntriesNum = len;
Expand Down
129 changes: 83 additions & 46 deletions gammu/src/gammu.c
Original file line number Diff line number Diff line change
Expand Up @@ -5725,7 +5725,7 @@ gammu_LinkSMS(PyObject *self, PyObject *args, PyObject *kwds)
static char *kwlist[] = {"Messages", "EMS", NULL};
GSM_Error error;
PyObject *value;
PyObject *ret;
PyObject *ret = NULL;
Py_ssize_t len;
int i;

Expand All @@ -5737,20 +5737,29 @@ gammu_LinkSMS(PyObject *self, PyObject *args, PyObject *kwds)

if (!MultiSMSListFromPython(value, &smsin)) return NULL;

smsout = (GSM_MultiSMSMessage **)malloc((len + 1) * sizeof(GSM_MultiSMSMessage *));
smsout = (GSM_MultiSMSMessage **)calloc((len + 1), sizeof(GSM_MultiSMSMessage *));
if (smsout == NULL) {
PyErr_NoMemory();
goto free_input;
}

error = GSM_LinkSMS(GSM_GetGlobalDebug(), smsin, smsout, ems);
if (!checkError(error, "LinkSMS")) return NULL;
if (!checkError(error, "LinkSMS")) {
ret = NULL;
goto free_output;
}

ret = MultiSMSListToPython(smsout);

free_output:
i = 0;
while(smsout[i] != NULL) {
while(i < len && smsout[i] != NULL) {
free(smsout[i]);
i++;
}
free(smsout);

free_input:
i = 0;
while(smsin[i] != NULL) {
free(smsin[i]);
Expand Down Expand Up @@ -5813,6 +5822,7 @@ gammu_EncodeSMS(PyObject *self, PyObject *args, PyObject *kwds)
{
GSM_MultiSMSMessage smsout;
GSM_MultiPartSMSInfo smsinfo;
GSM_Error error;
static char *kwlist[] = {"MessagesInfo", NULL};
PyObject *value;

Expand All @@ -5824,9 +5834,10 @@ gammu_EncodeSMS(PyObject *self, PyObject *args, PyObject *kwds)

if (!SMSInfoFromPython(value, &smsinfo)) return NULL;

if (!GSM_EncodeMultiPartSMS(GSM_GetGlobalDebug(), &smsinfo, &smsout)) {
error = GSM_EncodeMultiPartSMS(GSM_GetGlobalDebug(), &smsinfo, &smsout);
if (!checkError(error, "EncodeMultiPartSMS")) {
Comment on lines +5837 to +5838

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve the encoder's boolean return contract

With the Gammu 1.44.0 API used by the inspected workflows, GSM_EncodeMultiPartSMS returns gboolean, not GSM_Error, so assigning its result to error cannot propagate a specific encoding failure. A rejected oversized multipart input returns FALSE (0), which checkError converts to the generic GammuError rather than ERR_INVALIDDATA; consequently the newly added pytest.raises(gammu.ERR_INVALIDDATA) case fails, while successful calls only work accidentally because TRUE has the same numeric value as ERR_NONE.

Useful? React with 👍 / 👎.

GSM_FreeMultiPartSMSInfo(&smsinfo);
Py_RETURN_NONE;
return NULL;
}
GSM_FreeMultiPartSMSInfo(&smsinfo);

Expand Down Expand Up @@ -6485,8 +6496,10 @@ gammu_EncodePDU(PyObject *self, PyObject *args, PyObject *kwds)
GSM_Error error;
PyObject *value;
unsigned char buffer[1000];
char req[1000];
int length = 0, current = 0, i;
PyObject *result;
char *req;
int length = 0;
size_t current = 0, payload_length, smsc_length, number_length, result_length;
GSM_SMSMessage sms;
char *layout = NULL;
GSM_SMSMessageLayout *msg_layout;
Expand All @@ -6512,51 +6525,75 @@ gammu_EncodePDU(PyObject *self, PyObject *args, PyObject *kwds)
if (!checkError(error, "EncodeSMSFrame")) return NULL;

if (msg_layout == &PHONE_SMSDeliver) {
length = length - PHONE_SMSDeliver.Text;

for (i = 0;i < buffer[PHONE_SMSDeliver.SMSCNumber]+1;i++) {
req[current++]=buffer[PHONE_SMSDeliver.SMSCNumber+i];
}
req[current++]=buffer[PHONE_SMSDeliver.firstbyte];

for (i = 0;i<((buffer[PHONE_SMSDeliver.Number]+1)/2+1)+1;i++) {
req[current++]=buffer[PHONE_SMSDeliver.Number+i];
}
req[current++]=buffer[PHONE_SMSDeliver.TPPID];
req[current++]=buffer[PHONE_SMSDeliver.TPDCS];

for(i = 0;i < 7;i++) {
req[current++]=buffer[PHONE_SMSDeliver.DateTime+i];
if (length < PHONE_SMSDeliver.Text) {
checkError(ERR_INVALIDDATA, "EncodeSMSFrame");
return NULL;
}
req[current++]=buffer[PHONE_SMSDeliver.TPUDL];

for(i = 0;i < length;i++) {
req[current++]=buffer[PHONE_SMSDeliver.Text+i];
payload_length = (size_t)(length - PHONE_SMSDeliver.Text);
smsc_length = (size_t)buffer[PHONE_SMSDeliver.SMSCNumber] + 1;
number_length = ((size_t)buffer[PHONE_SMSDeliver.Number] + 1) / 2 + 2;
if ((size_t)PHONE_SMSDeliver.SMSCNumber + smsc_length > sizeof(buffer) ||
(size_t)PHONE_SMSDeliver.Number + number_length > sizeof(buffer) ||
payload_length > sizeof(buffer) - PHONE_SMSDeliver.Text) {
checkError(ERR_INVALIDDATA, "EncodeSMSFrame");
return NULL;
}
result_length = smsc_length + 1 + number_length + 2 + 7 + 1 + payload_length;
result = PyBytes_FromStringAndSize(NULL, (Py_ssize_t)result_length);
if (result == NULL) return NULL;
req = PyBytes_AS_STRING(result);
memcpy(req + current, buffer + PHONE_SMSDeliver.SMSCNumber, smsc_length);
current += smsc_length;
req[current++] = buffer[PHONE_SMSDeliver.firstbyte];
memcpy(req + current, buffer + PHONE_SMSDeliver.Number, number_length);
current += number_length;
req[current++] = buffer[PHONE_SMSDeliver.TPPID];
req[current++] = buffer[PHONE_SMSDeliver.TPDCS];
memcpy(req + current, buffer + PHONE_SMSDeliver.DateTime, 7);
current += 7;
req[current++] = buffer[PHONE_SMSDeliver.TPUDL];
memcpy(req + current, buffer + PHONE_SMSDeliver.Text, payload_length);
current += payload_length;
} else if (msg_layout == &PHONE_SMSSubmit) {
length = length - PHONE_SMSSubmit.Text;

for (i = 0;i < buffer[PHONE_SMSSubmit.SMSCNumber]+1;i++) {
req[current++]=buffer[PHONE_SMSSubmit.SMSCNumber+i];
}
req[current++]=buffer[PHONE_SMSSubmit.firstbyte];
req[current++]=buffer[PHONE_SMSSubmit.TPMR];

for (i = 0;i<((buffer[PHONE_SMSSubmit.Number]+1)/2+1)+1;i++) {
req[current++]=buffer[PHONE_SMSSubmit.Number+i];
if (length < PHONE_SMSSubmit.Text) {
checkError(ERR_INVALIDDATA, "EncodeSMSFrame");
return NULL;
}
req[current++]=buffer[PHONE_SMSSubmit.TPPID];
req[current++]=buffer[PHONE_SMSSubmit.TPDCS];
req[current++]=buffer[PHONE_SMSSubmit.TPVP];
req[current++]=buffer[PHONE_SMSSubmit.TPUDL];

for(i = 0;i < length;i++) {
req[current++]=buffer[PHONE_SMSSubmit.Text+i];
payload_length = (size_t)(length - PHONE_SMSSubmit.Text);
smsc_length = (size_t)buffer[PHONE_SMSSubmit.SMSCNumber] + 1;
number_length = ((size_t)buffer[PHONE_SMSSubmit.Number] + 1) / 2 + 2;
if ((size_t)PHONE_SMSSubmit.SMSCNumber + smsc_length > sizeof(buffer) ||
(size_t)PHONE_SMSSubmit.Number + number_length > sizeof(buffer) ||
payload_length > sizeof(buffer) - PHONE_SMSSubmit.Text) {
checkError(ERR_INVALIDDATA, "EncodeSMSFrame");
return NULL;
}
req[current+1]='\0';
result_length = smsc_length + 2 + number_length + 4 + payload_length;
result = PyBytes_FromStringAndSize(NULL, (Py_ssize_t)result_length);
if (result == NULL) return NULL;
req = PyBytes_AS_STRING(result);
memcpy(req + current, buffer + PHONE_SMSSubmit.SMSCNumber, smsc_length);
current += smsc_length;
req[current++] = buffer[PHONE_SMSSubmit.firstbyte];
req[current++] = buffer[PHONE_SMSSubmit.TPMR];
memcpy(req + current, buffer + PHONE_SMSSubmit.Number, number_length);
current += number_length;
req[current++] = buffer[PHONE_SMSSubmit.TPPID];
req[current++] = buffer[PHONE_SMSSubmit.TPDCS];
req[current++] = buffer[PHONE_SMSSubmit.TPVP];
req[current++] = buffer[PHONE_SMSSubmit.TPUDL];
memcpy(req + current, buffer + PHONE_SMSSubmit.Text, payload_length);
current += payload_length;
} else {
return PyBytes_FromStringAndSize("", 0);
}

return PyBytes_FromStringAndSize(req, current);
if (current != result_length) {
Py_DECREF(result);
PyErr_SetString(PyExc_RuntimeError, "Internal PDU length mismatch");
return NULL;
}
return result;
}

/* List of methods defined in the module */
Expand Down
50 changes: 50 additions & 0 deletions test/test_sms.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
import sys
import unittest

import pytest

import gammu

PDU_DATA = binascii.unhexlify(
Expand Down Expand Up @@ -65,6 +67,54 @@ def test_decode(self) -> None:
assert sms["Number"] == "604865888"
assert sms["Text"] == "Delivered"

def test_decode_rejects_truncated_frame(self) -> None:
with pytest.raises(gammu.ERR_CORRUPTED):
gammu.DecodePDU(b"")

def test_special_decoders_reject_malformed_data(self) -> None:
for value in (
"004000810004000000000000000807120500ffff0000",
"00400081000400000000000000100b0504158a00000003ce010130017f7f",
):
decoded = gammu.DecodePDU(bytes.fromhex(value))
assert gammu.DecodeSMS([decoded]) is None

def test_encode_pdu_rejects_extension_overflow(self) -> None:
sms = gammu.EncodeSMS(
{"Entries": [{"ID": "ConcatenatedTextLong", "Buffer": "^" * 80}]}
)[0]
sms["Text"] = "^" * 160
with pytest.raises(gammu.ERR_INVALIDDATA):
gammu.EncodePDU(sms)

def test_encode_rejects_unrepresentable_multipart(self) -> None:
with pytest.raises(gammu.ERR_INVALIDDATA):
gammu.EncodeSMS(
{"Entries": [{"ID": "ConcatenatedTextLong", "Buffer": "A" * 40000}]}
)

def test_conversion_limits_reject_instead_of_truncate(self) -> None:
sms = gammu.DecodePDU(
bytes.fromhex(
"0791361907001003B17A0C913619397750320000AD11CD701E340FB3C3F23CC81D0689C3BF"
)
)

oversized_binary = dict(sms, Coding="8bit", Text=b"A" * 651)
with pytest.raises(ValueError, match="SMS text is too large"):
gammu.EncodePDU(oversized_binary)

oversized_udh = dict(sms)
oversized_udh["UDH"] = {"Type": "UserUDH", "Text": b"\x00" * 141}
with pytest.raises(ValueError, match="UDH is too large"):
gammu.EncodePDU(oversized_udh)

with pytest.raises(ValueError, match="MultiSMS has too many entries"):
gammu.DecodeSMS([sms] * 51)

with pytest.raises(ValueError, match="Too many SMS info entries"):
gammu.EncodeSMS({"Entries": [{}] * 50})

def do_smstest(self, smsinfo, expected) -> None:
# encode SMSes
sms = gammu.EncodeSMS(smsinfo)
Expand Down
Loading