1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
#!/usr/bin/env python3
"""Example program that demonstrates the flow for retrieving a refresh token.
In order for this example to work your application's redirect URI must be set to
http://localhost:8080.
This tool can be used to conveniently create refresh tokens for later use with your web
application OAuth2 credentials.
"""
import asyncio
import os
import random
import socket
import sys
import asyncprawcore
def receive_connection():
"""Wait for and then return a connected socket..
Opens a TCP connection on port 8080, and waits for a single client.
"""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("localhost", 8080))
server.listen(1)
client = server.accept()[0]
server.close()
return client
def send_message(client, message):
"""Send message to client and close the connection."""
print(message)
client.send(f"HTTP/1.1 200 OK\r\n\r\n{message}".encode())
client.close()
async def main():
"""Provide the program's entry point when directly executed."""
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} SCOPE...")
return 1
requestor = asyncprawcore.Requestor("asyncprawcore_refresh_token_example")
try:
authenticator = asyncprawcore.TrustedAuthenticator(
requestor,
os.environ["PRAWCORE_CLIENT_ID"],
os.environ["PRAWCORE_CLIENT_SECRET"],
"http://localhost:8080",
)
state = str(random.randint(0, 65000)) # noqa: S311
url = authenticator.authorize_url("permanent", sys.argv[1:], state)
print(url)
client = receive_connection()
data = client.recv(1024).decode("utf-8")
param_tokens = data.split(" ", 2)[1].split("?", 1)[1].split("&")
params = dict([token.split("=") for token in param_tokens])
if state != params["state"]:
send_message(
client,
f"State mismatch. Expected: {state} Received: {params['state']}",
)
return 1
if "error" in params:
send_message(client, params["error"])
return 1
authorizer = asyncprawcore.Authorizer(authenticator)
await authorizer.authorize(params["code"])
send_message(client, f"Refresh token: {authorizer.refresh_token}")
finally:
await requestor.close()
return 0
if __name__ == "__main__":
sys.exit(asyncio.run(main()))
|
