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

profile photo of returntocorpreturntocorp
Author
unknown
Download Count*
LGPL Commons Clause
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

Semgrep CI docs

Defintion

Edit RuleView Source
Open in Playground 
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()