AppsTrafficController/cloudflare_api.py at main · ErikRoss/AppsTrafficController · GitHub

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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
import json
import requests


class CloudflareApi:
    def __init__(self):
        api_key = "0b00ffc31a16286878dce5211701287bf6bef"
        api_email = "nexodium@proton.me"
        api_token = "xkBpn4tp3HtNxmEXK4QPzpCYjaNx9MVF4OrQXTzA"
        account_id = "e8f5ac44c2848ca0c24e4c8bad429901"
        self.api_host = "https://api.cloudflare.com/client/v4"
        self.headers = {
            "X-Auth-Email": api_email,
            "X-Auth-Key": api_key,
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json",
        }
        self.params = {"account": {"id": account_id}}

    def create_zone(self, domain_name, zone_type="full"):
        endpoint = "/zones"
        url = self.api_host + endpoint
        data = {"name": domain_name, "type": zone_type}
        data.update(self.params)

        response = requests.post(url, headers=self.headers, data=json.dumps(data))

        if response.status_code != 200:
            result = response.json()["errors"][0]
            return {"success": False, "error": result}
        else:
            name_servers = response.json()["result"]["name_servers"]
            zone_id = response.json()["result"]["id"]
            return {"success": True, "nameservers": name_servers, "zone_id": zone_id}

    def get_zone(self, domain_name):
        endpoint = "/zones"
        url = self.api_host + endpoint
        params = {"name": domain_name}
        params.update(self.params)

        response = requests.get(url, headers=self.headers, params=params)
        if response.status_code != 200:
            result = response.json()["errors"][0]
            return {"error": result}
        else:
            result = response.json()["result"]
            print(result)
            nameservers = []
            zone_id = ""
            for zone in result:
                if zone["name"] == domain_name:
                    nameservers = zone["name_servers"]
                    zone_id = zone["id"]

            return {
                "success": True,
                "nameservers": nameservers,
                "zone_id": zone_id,
                "status": zone["status"]
                }

    def set_dns_records(self, zone_id: str, ip: str, name: str):
        endpoint = f"/zones/{zone_id}/dns_records"
        url = self.api_host + endpoint
        data = {
            "content": ip,
            "name": name,
            "proxied": True,
            "type": "A",
            "comment": "Created by AppsTrafficController",
            "ttl": 3600,
        }

        response = requests.post(url, headers=self.headers, data=json.dumps(data))
        if response.status_code != 200:
            result = response.json()["errors"][0]
            if result["code"] == 81057:
                return {"success": True, "result": result}
            return {"success": False, "error": result}
        else:
            result = response.json()["result"]
            return {"success": True, "result": result}

    def set_auto_https_rewriting(self, zone_id: str, state: str):
        endpoint = f"/zones/{zone_id}/settings/automatic_https_rewrites"
        url = self.api_host + endpoint
        data = {"value": state}

        response = requests.patch(url, headers=self.headers, data=json.dumps(data))
        if response.status_code != 200:
            result = response.json()["errors"][0]
            return {"error": result}
        else:
            result = response.json()["result"]["value"]
            return {"success": True, "result": result}

    def set_always_use_https(self, zone_id: str, state: str):
        endpoint = f"/zones/{zone_id}/settings/always_use_https"
        url = self.api_host + endpoint
        data = {"value": state}

        response = requests.patch(url, headers=self.headers, data=json.dumps(data))
        if response.status_code != 200:
            result = response.json()["errors"][0]
            return {"error": result}
        else:
            result = response.json()["result"]
            return {"success": True, "result": result}

    def get_managed_headers(self, zone_id: str):
        url = self.api_host + f"/zones/{zone_id}/managed_headers"
        response = requests.patch(url, headers=self.headers)
        if response.status_code != 200:
            result = response.json()["errors"][0]
            return {"error": result}
        else:
            result = response.json()["result"]
            return {"success": True, "result": result}

    def add_visitor_location_header(self, zone_id: str):
        endpoint = f"/zones/{zone_id}/managed_headers"
        url = self.api_host + endpoint
        data = {
            "managed_request_headers": [
                {"id": "add_visitor_location_headers", "enabled": True}
            ]
        }

        response = requests.patch(url, headers=self.headers, data=json.dumps(data))
        if response.status_code != 200:
            result = response.json()["errors"][0]
            return {"error": result}
        else:
            result = response.json()["result"]
            return {"success": True, "result": result}