python.correctness.socket-shutdown-close.socket-shutdown-close

Author
unknown
Download Count*
License
Socket is not closed if shutdown fails. When socket.shutdown fails on an OSError, socket.close is not called and the code fails to clean up the socket and allow garbage collection to release the memory used for it. The OSError on shutdown can occur when the remote side of the connection closes the connection first.
Run Locally
Run in CI
Defintion
rules:
- id: socket-shutdown-close
patterns:
- pattern: |
$SOCK.shutdown(socket.$A)
$SOCK.close()
- pattern-not-inside: |
try:
...
except ...:
...
$SOCK.close()
- pattern-not-inside: |
try:
...
finally:
...
$SOCK.close()
message: Socket is not closed if shutdown fails. When socket.shutdown fails on
an OSError, socket.close is not called and the code fails to clean up the
socket and allow garbage collection to release the memory used for it. The
OSError on shutdown can occur when the remote side of the connection
closes the connection first.
metadata:
references:
- https://www.paulsprogrammingnotes.com/2021/12/python-memory-leaks.html
category: correctness
technology:
- python
license: Commons Clause License Condition v1.0[LGPL-2.1-only]
languages:
- python
severity: WARNING
Examples
socket-shutdown-close.py
# From https://www.paulsprogrammingnotes.com/2021/12/python-memory-leaks.html
# naive case
sock = socket.socket(af, socktype, proto)
try:
# ruleid: socket-shutdown-close
sock.shutdown(socket.SHUT_RDWR)
sock.close()
except OSError:
pass
try:
# ok: socket-shutdown-close
sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
try:
sock.close()
except OSError:
pass
# other variable name
other_name = socket.socket(af, socktype, proto)
try:
# ruleid: socket-shutdown-close
other_name.shutdown(socket.SHUT_RDWR)
other_name.close()
except OSError:
pass
try:
# ok: socket-shutdown-close
other_name.shutdown(socket.SHUT_RDWR)
except OSError:
pass
try:
other_name.close()
except OSError:
pass
# different argument to shutdown(...)
from socket import SHUT_WR
sock = socket.socket(af, socktype, proto)
try:
# ruleid: socket-shutdown-close
sock.shutdown(SHUT_WR)
sock.close()
except OSError:
pass
try:
# ok: socket-shutdown-close
sock.shutdown(SHUT_WR)
except OSError:
pass
try:
sock.close()
except OSError:
pass
# non-socket object
# ok: socket-shutdown-close
logging.shutdown()
logging.close()
# close is called after all
sock = socket.socket(af, socktype, proto)
try:
# ok: socket-shutdown-close
sock.shutdown(socket.SHUT_RDWR)
sock.close()
except OSError:
sock.close()
try:
# ok: socket-shutdown-close
sock.shutdown(socket.SHUT_RDWR)
sock.close()
pass
except OSError:
pass
sock.close()
pass
try:
# ok: socket-shutdown-close
sock.shutdown(socket.SHUT_RDWR)
sock.close()
except:
sock.close()
try:
# ok: socket-shutdown-close
sock.shutdown(socket.SHUT_RDWR)
sock.close()
finally:
sock.close()
Short Link: https://sg.run/Bel5