· 5 years ago · May 20, 2020, 01:38 PM
1"""CDN Origin Switcher is a simple tool to connect to CDN's API and switch the origin in an even of DR"""
2import argparse
3import socket
4import sys
5import requests
6from requests.auth import HTTPBasicAuth
7
8# CDN API URI
9SITES_URI = "/v1/content/sites"
10
11
12# TODO
13# here are a lot of copy/paste of
14# resp = requests.put(f"https://{args.lb_ip}{SITES_URI}/{site_name}/edit",
15# headers=headers,
16# auth=HTTPBasicAuth(
17# username=args.user,
18# password=args.password
19# ),
20# data=site_conf,
21# verify=False)
22# it makes sense to use requests.Session() here, init it once and use everywhere
23# 3.It's better to use logging lib instead of prints. Prints are usually used just for temporary debug lines,
24# not in prod code
25
26def create_parser() -> argparse.ArgumentParser:
27 """
28 This function is called from main() and return a ArgumentParser object
29 :return: return ArgumentParser object
30 """
31 parser = argparse.ArgumentParser(description="LGI DNS Origin Switcher")
32 parser.add_argument("-i", "--load-balancer-ip-address", dest="lb_ip", required=True,
33 help="IP address of the load balancer which will be called for switchover")
34 parser.add_argument("-u", "--user", dest="user", required=True,
35 help="Load Balancer admin user")
36 parser.add_argument("-p", "--password", dest="password", required=True,
37 help="Load Balancer admin password")
38 parser.add_argument("-n", "--load-balancer-hostname", dest="lb_host", required=True,
39 help="Load Balancer hostname")
40 parser.add_argument("-s", "--site-name", dest="site", required=True,
41 help="Site name to switch")
42 parser.add_argument("-o", "--new-origin", dest="origin",
43 help="New origin for CDN site followed by comma (192.168.1.1,192.168.1.2,...)")
44 parser.add_argument("-a", "--aws-load-balancer", dest="aws_lb",
45 help="AWS Load Balancer DNS name to look up for IPs")
46 # You either supply AWS LB hostname or a set of origins manually
47 # TODO Fix this shit
48 # gr = parser.add_mutually_exclusive_group()
49 # gr.add_argument("-a", "--aws-load-balancer")
50 # gr.add_argument("-o", "--new-origin")
51 return parser
52
53
54def obtain_arguments(parser: argparse.ArgumentParser) -> argparse.Namespace:
55 """
56 Parse arguments and return arguments objects to be used by other functions
57 :param parser: parser object
58 :return: return argparse.Namespace object
59 """
60 # TODO not critical but seems as an overhead to wrap one line into a function
61 return parser.parse_args()
62
63
64def check_ip_addr_validity(ip: str) -> bool:
65 """
66 Checks if provided IP address is a valid IP address
67 :param ip: IP addr from CLI argument
68 :return: True or False
69 """
70 try:
71 # TODO ipaddress lib validates IPs better
72 # f.e. socket.inet_aton("1") -> b'\x00\x00\x00\x01' (passed, but in fact IP is incorrect)
73 # ipaddress.ip_address('1') -> error
74 socket.inet_aton(ip)
75 return True
76 except socket.error:
77 return False
78
79
80# TODO IMHO it will be more clear to pass only the arguments which are used in the function, not pass all args namespace
81def get_site_name(args: argparse.Namespace) -> str:
82 """
83 Connects to the CDN API and gets the site name
84 :param args: arguments object from CLI
85 :return: site name (string)
86 """
87 site_name = ""
88 headers = {"Host": str(args.lb_host)}
89 resp = requests.get(f"https://{args.lb_ip}{SITES_URI}",
90 headers=headers,
91 auth=HTTPBasicAuth(
92 username=args.user,
93 password=args.password
94 ),
95 verify=False)
96 if resp.status_code != 200:
97 print(f"Request failed with code {resp.status_code}: {resp.text}")
98 else:
99 sites = resp.json()["_links"]["http://uri.velocix.com/relation/lists"]
100 for site in sites:
101 if site['title'] == args.site:
102 site_name = site['name']
103 break
104 return site_name
105
106
107def get_site_etag(args: argparse.Namespace) -> str:
108 """
109 Connect to CDN API and gets site's ETag
110 :param args: arguments object from CLI
111 :return: site ETag (string)
112 """
113 etag = ""
114 headers = {"Host": str(args.lb_host)}
115 site_name = get_site_name(args)
116 resp = requests.get(f"https://{args.lb_ip}{SITES_URI}/{site_name}/edit",
117 headers=headers,
118 auth=HTTPBasicAuth(
119 username=args.user,
120 password=args.password
121 ),
122 verify=False)
123 if resp.status_code != 200:
124 print(f"Request failed with code {resp.status_code}: {resp.text}")
125 else:
126 etag = resp.headers['etag']
127 return etag
128
129
130def get_site_configuration(args: argparse.Namespace) -> dict:
131 """
132 Connect to CDN API and obtain site configuration in a JSON format
133 :param args: arguments object from CLI
134 :return: Site configuration dict
135 """
136 site_conf = {}
137 headers = {"Host": str(args.lb_host)}
138 site_name = get_site_name(args)
139 resp = requests.get(f"https://{args.lb_ip}{SITES_URI}/{site_name}",
140 headers=headers,
141 auth=HTTPBasicAuth(
142 username=args.user,
143 password=args.password
144 ),
145 verify=False)
146 if resp.status_code != 200:
147 print(f"Request failed with code {resp.status_code}: {resp.text}")
148 else:
149 site_conf = resp.json()
150 return site_conf
151
152
153def get_aws_lb_ips(args: argparse.Namespace) -> [str]: # TODO incorrect annotation, please use List[str]
154 """
155 Query AWS LB's DNS name and obtain resolved IP addresses
156 :param args: arguments object from CLI
157 :return: list of IP addresses
158 """
159 ips = [] # TODO the var is not used anywhere
160 # socket.gethostbyname_ex return tuple, where 3rd element is a list of IP addresses
161 ips = socket.gethostbyname_ex(args.aws_lb)[2]
162 return ips
163
164
165# TODO add ports for origins
166def set_site_origin(args: argparse.Namespace) -> int:
167 """
168 Connect to CDN API and replace origin servers with the ones provided in CLI
169 :param args: arguments object from CLI
170 :return: HTTP return code
171 """
172
173 site_conf = get_site_configuration(args)
174
175 if "originServers" not in site_conf:
176 print("Cannot find originServers key in the site configuration object!")
177 return 404
178
179 if args.origin:
180 site_conf['originServers'] = args.origin.split(",")
181 elif args.aws_lb:
182 origin_servers = get_aws_lb_ips(args)
183 if origin_servers:
184 site_conf['originServers'] = origin_servers
185 else:
186 print("Cannot obtain AWS LB IP addresses")
187 return 404
188 else:
189 print("No new origin provided! Nothing to be changed")
190
191 site_name = get_site_name(args)
192
193 headers = {"Host": str(args.lb_host),
194 "If-Match": get_site_etag(args)}
195
196 resp = requests.put(f"https://{args.lb_ip}{SITES_URI}/{site_name}/edit",
197 headers=headers,
198 auth=HTTPBasicAuth(
199 username=args.user,
200 password=args.password
201 ),
202 data=site_conf,
203 verify=False)
204 return_code = int(resp.status_code)
205 return return_code
206
207
208# TODO add ports to origins
209def main():
210 """Main"""
211 parser = create_parser()
212 args = obtain_arguments(parser)
213
214 if not check_ip_addr_validity(args.lb_ip):
215 print(f"Provided IP address {args.lb_ip} is not valid!")
216 sys.exit(1)
217
218 if args.origin:
219 for origin in args.origin.split(','):
220 if not check_ip_addr_validity(origin):
221 print(f"Origin IP {origin} failed the validity check! Can not proceed!")
222 sys.exit(1)
223 rc = set_site_origin(args)
224 print(rc)
225
226
227if __name__ == "__main__":
228 main()