Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_long(self):
sock = socket.socket()
try:
select.select(
[six.builtins.long(sock.fileno())], [], [], 0.001)
finally:
sock.close()
def _create_connection(self):
conn = socket.socket()
self._close_on_teardown(conn)
conn.connect(('127.0.0.1', self._server.server_port))
return conn
def makefile(self, timeout=0.1, bufsize=1):
sock = socket.socket()
sock.connect((self.server.server_host, self.server.server_port))
fobj = sock.makefile(bufsize=bufsize)
fobj._sock.settimeout(timeout)
sock.close()
return fobj
def recv(self, buflen=1024, flags=0):
self._checkClosed()
if self._sslobj:
if flags != 0:
raise ValueError(
"non-zero flags not allowed in calls to recv() on %s" %
self.__class__)
if buflen == 0:
# https://github.com/python/cpython/commit/00915577dd84ba75016400793bf547666e6b29b5
# Python #23804
return b''
return self.read(buflen)
else:
return socket.recv(self, buflen, flags)
def _check_send_connection(self):
if self._push_connection is None:
tcp_socket = socket.socket(
socket.AF_INET, socket.SOCK_STREAM, 0)
s = ssl.wrap_socket(tcp_socket, ssl_version=ssl.PROTOCOL_TLSv1,
**self._sslargs)
addr = ["gateway.push.apple.com", 2195]
if self._sandbox:
addr[0] = "gateway.sandbox.push.apple.com"
logger.debug('Connecting to %s' % addr[0])
s.connect_ex(tuple(addr))
self._push_connection = s
self._error_greenlet = gevent.spawn(self.save_err,
self._error_loop)
def __init__(self, host=None, port=None, app=None):
self.host, self.port, self.app = host, port, app
self.logger = self.app.logger
self.crypto = crypto_pool.alloc(self.app.config.crypto_no)
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.host, self.port))
if "ip" not in self.app.config:
self.app.config["ip"] = "0.0.0.0" # server allocate address from pull (like simple dummy DHCP)
super(VPNServerConnection, self).__init__()
self.make_handshake()
def accept(self):
sock, addr = socket.accept(self)
client = SSLObject(sock._sock, server_side=True)
client.do_handshake()
return client, addr
def recvfrom(self, buflen=1024, flags=0):
self._checkClosed()
if self._sslobj:
raise ValueError("recvfrom not allowed on instances of %s" %
self.__class__)
return socket.recvfrom(self, buflen, flags)
# gevent-test-requires-resource: unixsocket_client
import os
from gevent.pywsgi import WSGIServer
from gevent import socket
def application(environ, start_response):
assert environ
start_response('200 OK', [])
return []
if __name__ == '__main__':
listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sockname = './' + os.path.basename(__file__) + '.sock'
if os.path.exists(sockname):
os.remove(sockname)
listener.bind(sockname)
listener.listen(1)
WSGIServer(listener, application).serve_forever()
import struct
import time
import sqlite3
# -- third party --
from gevent.event import Event
from gevent.pool import Pool
from gevent.socket import socket as gsock
import gevent
import requests
import spidermonkey
# -- own --
# -- code --
if gsock is not socket:
from gevent import monkey
monkey.patch_all()
log = logging.getLogger('QQBot')
UA_STRING = (
'Mozilla/5.0 (X11; Linux x86_64) '
'AppleWebKit/537.36 (KHTML, like Gecko) '
'Ubuntu Chromium/49.0.2623.108 '
'Chrome/49.0.2623.108 Safari/537.36'
)
def naive_qs(d):
return '&'.join(['%s=%s' % (k, v) for k, v in d.iteritems()])