xan/xenu_nntp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prevent passing wrong number of args to POST

Browse source
This commit is contained in:
XANTRONIX Development 2024-11-30 19:57:29 -05:00
parent 61f26dd102
commit c2b2181c7f
1 changed files with 35 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

84
lib/nntp/tiny/session.py
View file

@ -740,11 +740,33 @@ class Session():
return True return True
def _cmd_post(self): def _post_impl(self, message_id=None):
if self.perms is None or not self.perms & UserPermission.POST: if self.perms is None or not self.perms & UserPermission.POST:
return self.respond(ResponseCode.NNTP_POST_PROHIBITED) return self.respond(ResponseCode.NNTP_POST_PROHIBITED)
self.respond(ResponseCode.NNTP_INQUIRY_ARTICLE) if message_id:
sql = """
select
count(message_id)
from
message
where
message_id = ?
"""
cr = self.db.execute(sql, (message_id,))
row = cr.fetchone()
if row is not None and row[0] > 0:
return self.respond(ResponseCode.NNTP_ARTICLE_NOT_WANTED_ID)
code_inquiry = ResponseCode.NNTP_INQUIRY_ARTICLE_ID
code_received = ResponseCode.NNTP_ARTICLE_RECEIVED_ID
else:
code_inquiry = ResponseCode.NNTP_INQUIRY_ARTICLE
code_received = ResponseCode.NNTP_ARTICLE_RECEIVED
self.respond(code_inquiry)
message = Message() message = Message()
@ -754,63 +776,27 @@ class Session():
if line == '': if line == '':
self.active = False self.active = False
break break
elif line == '.':
stripped = line.rstrip()
if stripped == '.':
if self._save_message(message): if self._save_message(message):
return self.respond(ResponseCode.NNTP_ARTICLE_RECEIVED) return self.respond(code_received)
else: else:
return self.respond(ResponseCode.NNTP_POST_FAILED) return self.respond(ResponseCode.NNTP_POST_FAILED)
elif line == '..': elif stripped == '..':
line = '.' line = line[1:]
try: try:
message.readline(line) message.readline(line)
except: except:
return self.respond(ResponseCode.NNTP_POST_FAILED) return self.respond(ResponseCode.NNTP_POST_FAILED)
def _cmd_post(self):
return self._post_impl()
def _cmd_ihave(self, message_id): def _cmd_ihave(self, message_id):
if self.perms is None or not self.perms & UserPermission.POST: return self._post_impl(message_id)
return self.respond(ResponseCode.NNTP_POST_PROHIBITED)
sql = """
select
count(message_id)
from
message
where
message_id = ?
"""
cr = self.db.execute(sql, (message_id,))
row = cr.fetchone()
if row is not None and row[0] > 0:
return self.respond(ResponseCode.NNTP_ARTICLE_NOT_WANTED_ID)
self.respond(ResponseCode.NNTP_INQUIRY_ARTICLE_ID)
message = Message()
while True:
line = self.readline()
if line == '':
self.active = False
break
elif line == '.':
if message.message_id != message_id:
return self.respond(ResponseCode.NNTP_POST_FAILED)
if self._save_message(message):
return self.respond(ResponseCode.NNTP_ARTICLE_RECEIVED_ID)
else:
return self.respond(ResponseCode.NNTP_POST_FAILED)
elif line == '..':
line = '.'
try:
message.readline(line)
except:
return self.respond(ResponseCode.NNTP_POST_FAILED)
def _cmd_date(self): def _cmd_date(self):
timestamp = datetime.datetime.now(datetime.UTC) timestamp = datetime.datetime.now(datetime.UTC)