dnspod.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import requests
import sys


class DnsPod():
    def __init__(self, user, password):
        self.__user = user
        self.__password = password

    def request(self, url, data):
        data['format'] = 'json'
        data['login_email'] = self.__user
        data['login_password'] = self.__password
        result = requests.post(url, data=data)
        return result.json()

    def list_domains(self):
        url = 'https://dnsapi.cn/Domain.List'
        data = {}
        result = self.request(url, data)
        if 'domains' in result:
            return result['domains']
        else:
            return []

    def show_domain(self, domain):
        url = 'https://dnsapi.cn/Domain.Info'
        data = {'domain': domain}
        result = self.request(url, data)
        if 'domain' in result:
            return result['domain']
        else:
            return {}

    def list_records(self, domain, sub_domain=None, keyword=None):
        domain = self.show_domain(domain)
        if not domain:
            return []
        url = 'https://dnsapi.cn/Record.List'
        data = {
            'domain_id': domain['id']
        }
        if sub_domain is not None:
            data['sub_domain'] = sub_domain
        if keyword is not None:
            data['keyword'] = keyword
        result = self.request(url, data)
        if 'records' in result:
            return result['records']
        else:
            return []

    def create_record(self, domain, name, type, line, value, mx, ttl, enable):
        domain = self.show_domain(domain)
        if not domain:
            raise ValueError('Bad domain name')
        url = 'https://dnsapi.cn/Record.Create'
        data = {
            'domain_id': domain['id'],
            'sub_domain': name,
            'record_type': type,
            'record_line': line,
            'value': value,
            'mx': mx,
            'ttl': ttl,
            'status': 'enable' if enable else 'disable'
        }
        result = self.request(url, data)
        if result['status']['code'] != '1':
            raise ValueError(result['status']['message'])
        return result['record']

    def update_record(self, domain, id, name, type, line, value, mx, ttl, enable):
        domain = self.show_domain(domain)
        if not domain:
            raise ValueError('Bad domain name')
        url = 'https://dnsapi.cn/Record.Modify'
        data = {
            'domain_id': domain['id'],
            'record_id': id,
            'sub_domain': name,
            'record_type': type,
            'record_line': line,
            'value': value,
            'mx': mx,
            'ttl': ttl,
            'status': 'enable' if enable else 'disable'
        }
        result = self.request(url, data)
        if result['status']['code'] != '1':
            raise ValueError(result['status']['message'])
        return result['record']

    def delete_record(self, domain, id):
        domain = self.show_domain(domain)
        if not domain:
            raise ValueError('Bad domain name')
        url = 'https://dnsapi.cn/Record.Remove'
        data = {
            'domain_id': domain['id'],
            'record_id': id
        }
        result = self.request(url, data)
        if result['status']['code'] != '1':
            raise ValueError(result['status']['message'])


def acme_challenge(name, value):
    client = DnsPod('YourEmail', 'YourPassword')
    domain = 'YourDomain'
    name = '_acme-challenge.' + name
    name = name[:len(name) - len(domain) - 1]
    records = client.list_records(domain, sub_domain=name)
    if records and (len(records) > 1 or records[0]['type'] != 'TXT'):
        raise ValueError('Duplicate or invalid _acme-challenge records found')
    if records:
        id = records[0]['id']
        client.update_record(domain, id, name, 'TXT', '默认', value, 0, 600, True)
    else:
        client.create_record(domain, name, 'TXT', '默认', value, 0, 600, True)


if __name__ == '__main__':
    if len(sys.argv) != 3:
        print('usage: %s domain value' % sys.argv[0])
        sys.exit(1)
    name = sys.argv[1]
    value = sys.argv[2]
    print('acme_challenge: %s' % name)
    acme_challenge(name, value)
    print('done')