我怎么做这个?
这是我的密码
import websockets
async def test():
async with websockets.connect('ws://iqoption.com') as websocket:
response = await websocket.recv()
print(response)
# Client async code
问题是,我可以发送这个头来在服务器上进行身份验证吗?
标头
'Host':'iqoption.com',
'User-Agent': 'Mozilla/5.0 (Windows NT 6.3; Win64; x64; rv:73.0) Gecko/20100101 Firefox/73.0',
'Accept' : '*/*',
'Accept-Encoding': 'gzip, deflate',
'Sec-WebSocket-Version' : '13',
'Origin' : 'https://iqoption.com',
'Sec-WebSocket-Key': 'iExBWv1j6sC1uee2qD+QPQ==',
'Connection' : 'keep-alive, Upgrade',
'Upgrade': 'websocket'}
我可以用其他代码来做,但是这些消息仍然是用tls加密的。
proxies = {"http": "http://127.0.0.1:8080", "https": "http://127.0.0.1:8080"}
urllib3.disable_warnings()
r = requests.get('https://iqoption.com/echo/websocket', stream=True,headers = Headers, proxies=proxies, verify=False)
#r = requests.get('https://iqoption.com/echo/websocket', stream=True,headers = Headers)
s = socket.fromfd(r.raw.fileno(), socket.AF_INET, socket.SOCK_STREAM)
def receive_and_print():
#for message in iter(lambda: s.recv(1024).decode("utf-8", errors="replace"), ''):
for message in iter(lambda: s.recv(1024).decode( errors="replace"), ''):
print(":", message)
print("")
import threading
background_thread = threading.Thread(target=receive_and_print)
background_thread.daemon = True
background_thread.start()
while 1:
s.send(input("Please enter your message: ").encode())
print("Sent")
print("")
有什么建议吗?
发布于 2020-02-19 22:37:17
我认为您目前缺少对WebSockets的基本理解,正如您以前的实验所示。WebSockets不是普通的套接字。WebSockets是在HTTP握手之后创建的一些类似套接字的想法。您不能像在requests
之后尝试那样直接从连接中获取套接字,但是您必须明确地使用在RFC 6455中定义的WebSocket应用程序协议。
也就是说,使用像websockets
这样的库要好得多。但是您仍然必须使用实际的WebSocket端点作为目标,而不仅仅是在同一台服务器上使用一些任意的URL。在这种情况下,端点不是ws://iqoption.com
,而是wss://iqoption.com/echo/websocket
,即更长的路径和wss://
,而不是ws://
。
如果没有正确的URL,您就会得到错误消息,您似乎会将其解释为身份验证问题。但是这里根本不涉及身份验证。一旦使用了适当的端点,它就可以正常工作,而不需要发送特定的头:
async def test():
async with websockets.connect("wss://iqoption.com/echo/websocket") as websocket:
response = await websocket.recv()
print(response)
发布于 2022-11-18 03:42:50
我认为您可以直接在请求中添加标题。下面是我使用websockets
的例子。
async def hello(uri):
async with websockets.connect(uri=uri, extra_headers= {'Auth':'yourTokenHere'}) as websocket:
await websocket.send(json.dumps({"action":"hello", "message":"hi"})) # AWS.APIGateway style for example
message = await websocket.recv()
https://stackoverflow.com/questions/60308749
复制相似问题