forked from schustudios/SimplePush
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlambda_function.py
More file actions
executable file
·113 lines (85 loc) · 2.92 KB
/
Copy pathlambda_function.py
File metadata and controls
executable file
·113 lines (85 loc) · 2.92 KB
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
#!/usr/bin/env python3
from hyper import HTTPConnection, HTTP20Connection
from hyper.tls import init_context
import json
devEnvironments = ["dev", "development"]
prodEnvironments = ["prod", "production"]
def getOrRaise(dict, key, default=None):
result = dict.get(key, default)
if result == None:
raise ValueError("Missing {} key in request".format(key))
return result
def getEnvironment(dict):
result = getOrRaise(dict, "environment")
if result in devEnvironments + prodEnvironments:
return result
raise ValueError("Invalid Environment found. Please set either 'dev' or 'prod' as the value for 'environment' key in request.")
def send_push(input):
topic = getOrRaise(input, "topic")
token = getOrRaise(input, "token_hex")
apnsPayload = getOrRaise(input, "apns")
environment = getEnvironment(input)
if environment in devEnvironments:
host = "api.development.push.apple.com"
pushCert = "cert/pushcert_dev.p12"
if environment in prodEnvironments:
host = "api.push.apple.com"
pushCert = "cert/pushcert_prod.p12"
print(host)
method = "POST"
path = "/3/device/{}".format(token)
# Build headers
headers = {"apns-topic": topic}
if input.get("apns-push-type"):
headers["apns-push-type"] = input.get("apns-push-type")
if input.get("apns-id"):
headers["apns-id"] = input.get("apns-id")
if input.get("apns-expiration"):
headers["apns-expiration"] = input.get("apns-expiration")
if input.get("apns-priority"):
headers["apns-priority"] = input.get("apns-priority")
if input.get("apns-collapse-id"):
headers["apns-collapse-id"] = input.get("apns-collapse-id")
conn = HTTPConnection(
host=host,
secure=True,
port=443,
ssl_context=init_context(cert=pushCert)
)
conn.request(
method=method,
url=path,
body=json.dumps(apnsPayload).encode("utf-8"),
headers=headers
)
return conn.get_response()
def lambda_handler(event, context):
try:
response = send_push(event)
return {
"statusCode": response.status,
"body": response.read().decode("utf-8")
}
except Exception as e:
return {
"statusCode": 400,
"body": str(e)
}
if __name__ == "__main__":
# Running Executable from Script
apnsPayload = {
"aps": { "alert": "Test Alert!" },
"acme": "Some Test Metadata"
}
event = {
"topic": "com.schustudios.PushExample",
"token_hex": "4177cc6b62437e0943c5b9fd3d5bad8344c80ae7ed38bb429d7bdb26b1a6f042",
"environment": "development",
"apns": apnsPayload
}
try:
response = send_push(event)
print("Status Code: {}".format(response.status))
print(response.read().decode("utf-8"))
except Exception as e:
print(e)