CDN:Use the EdgeScript CLI to manage scripts

Create the project files

1. Create a folder for the project, for example, cdn-api.

2. Inside the folder, create a file named aliyun.ini to store your RAM user credentials:

   [Credentials]
   accesskeyid = <your-access-key-id>
   accesskeysecret = <your-access-key-secret>

3. Inside the same folder, create a file named es.py and paste in the following Python script:

   #!/usr/bin/python
   # -*- coding:utf-8 -*-
   
   import sys, os
   import urllib, urllib2
   import base64
   import hmac
   import hashlib
   from hashlib import sha1
   import time
   import uuid
   import json
   from optparse import OptionParser
   import ConfigParser
   import traceback
   import urllib2
   
   access_key_id = '';
   access_key_secret = '';
   cdn_server_address = 'https://cdn.aliyuncs.com'
   CONFIGFILE = os.getcwd() + '/aliyun.ini'
   CONFIGSECTION = 'Credentials'
   cmdlist = '''
   
      1. Publish the ES rule to the simulated environment or production environment
         ./es.py action=push_test_env domain=<domain> rule='{"pos":"<head|foot>","pri":"0-999","rule_path":"<the es code path>","enable":"<on|off>"}'
         ./es.py action=push_product_env domain=<domain> rule='{"pos":"<head|foot>","pri":"0-999","rule_path":"<the es code path>","enable":"<on|off>","configid":"<configid>"}'
   
      2. Query the ES rule in the simulated environment or production environment
         ./es.py action=query_test_env domain=<domain>
         ./es.py action=query_product_env domain=<domain>
   
      3. Delete the ES rule in the simulated environment or production environment
         ./es.py action=del_test_env domain=<domain> configid=<configid>
         ./es.py action=del_product_env domain=<domain> configid=<configid>
   
      4. Publish the ES rule from the simulated to production environment, or Rollback the ES rule in the simulated environment
         ./es.py action=publish_test_env domain=<domain>
         ./es.py action=rollback_test_env domain=<domain>
   '''
   
   def percent_encode(str):
       res = urllib.quote(str.decode(sys.stdin.encoding).encode('utf8'), '')
       res = res.replace('+', '%20')
       res = res.replace('*', '%2A')
       res = res.replace('%7E', '~')
       return res
   
   def compute_signature(parameters, access_key_secret):
       sortedParameters = sorted(parameters.items(), key=lambda parameters: parameters[0])
   
       canonicalizedQueryString = ''
       for (k,v) in sortedParameters:
           canonicalizedQueryString += '&' + percent_encode(k) + '=' + percent_encode(v)
   
       stringToSign = 'GET&%2F&' + percent_encode(canonicalizedQueryString[1:])
   
       h = hmac.new(access_key_secret + "&", stringToSign, sha1)
       signature = base64.encodestring(h.digest()).strip()
       return signature
   
   def compose_url(user_params):
       timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
   
       parameters = { \
           'Format'        : 'JSON', \
           'Version'       : '2018-05-10', \
           'AccessKeyId'   : access_key_id, \
           'SignatureVersion'  : '1.0', \
           'SignatureMethod'   : 'HMAC-SHA1', \
           'SignatureNonce'    : str(uuid.uuid1()), \
           'Timestamp'         : timestamp, \
       }
   
       for key in user_params.keys():
           parameters[key] = user_params[key]
   
       signature = compute_signature(parameters, access_key_secret)
       parameters['Signature'] = signature
       url = cdn_server_address + "/?" + urllib.urlencode(parameters)
       return url
   
   def make_request(user_params, quiet=False):
       url = compose_url(user_params)
   
       try:
           req = urllib2.Request(url)
           r = urllib2.urlopen(req)
       except urllib2.HTTPError, err:
           print "Response Code:\n============="
           print err
           body=err.read()
           body_json = json.loads(body)
           body_str =json.dumps(body_json,indent=4)
           print "\nResponse Info:\n=============="
           print body_str
           return
   
       if r.getcode() == 200:
           print "Response Code:\n=============\n200 OK"
       print "\nResponse Info:\n=============="
       body=r.read()
       body_json = json.loads(body)
       body_str =json.dumps(body_json,indent=4)
       print body_str
   
   def configure_accesskeypair(args, options):
       if options.accesskeyid is None or options.accesskeysecret is None:
           print("config miss parameters, use --id=[accesskeyid] --secret=[accesskeysecret]")
           sys.exit(1)
       config = ConfigParser.RawConfigParser()
       config.add_section(CONFIGSECTION)
       config.set(CONFIGSECTION, 'accesskeyid', options.accesskeyid)
       config.set(CONFIGSECTION, 'accesskeysecret', options.accesskeysecret)
       cfgfile = open(CONFIGFILE, 'w+')
       config.write(cfgfile)
       cfgfile.close()
   
   def setup_credentials():
       config = ConfigParser.ConfigParser()
       try:
           config.read(CONFIGFILE)
           global access_key_id
           global access_key_secret
           access_key_id = config.get(CONFIGSECTION, 'accesskeyid')
           access_key_secret = config.get(CONFIGSECTION, 'accesskeysecret')
       except Exception, e:
           print traceback.format_exc()
           print("can't get access key pair, use config --id=[accesskeyid] --secret=[accesskeysecret] to setup")
           sys.exit(1)
   
   def parse_args(user_params):
       req_args = {}
   
       if user_params['action'] == 'push_test_env' or user_params['action'] == 'push_product_env':
           if not user_params.has_key('domain') or not user_params.has_key('rule'):
               parser.print_help()
               sys.exit(0)
   
           data = []
           for rule in user_params['rule']:
               rule_cfg = {
                   'functionId' : 180,
                   'functionName' : 'edge_function',
                   'functionArgs' : []
               }
               for k in rule:
                   arg_cfg = {}
                   if k == 'configid':
                       rule_cfg['configId'] = int(rule[k])
                   elif k == 'rule_path':
                       try:
                           f = open(rule[k], "r")
                           code = f.read()
                       except IOError:
                           print "io error"
                           sys.exit(0)
                       else:
                           f.close()
                       arg_cfg['argName'] = 'rule'
                       arg_cfg['argValue'] = code
                       rule_cfg['functionArgs'].append(arg_cfg)
                   else:
                       arg_cfg['argName'] = k
                       arg_cfg['argValue'] = rule[k]
                       rule_cfg['functionArgs'].append(arg_cfg)
               data.append(rule_cfg)
           rule_str = json.dumps(data)
   
           if user_params['action'] == 'push_test_env':
               req_args = {'Action':'SetCdnDomainStagingConfig', 'DomainName':user_params['domain'], 'Functions':rule_str}
           else:
               req_args = {'Action':'BatchSetCdnDomainConfig', 'DomainNames':user_params['domain'], 'Functions':rule_str}
   
       elif user_params['action'] == 'query_test_env':
           if not user_params.has_key('domain'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'DescribeCdnDomainStagingConfig', 'DomainName':user_params['domain'], 'FunctionNames':'edge_function'}
   
       elif user_params['action'] == 'query_product_env':
           if not user_params.has_key('domain'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'DescribeCdnDomainConfigs', 'DomainName':user_params['domain'], 'FunctionNames':'edge_function'}
   
       elif user_params['action'] == 'del_test_env':
           if not user_params.has_key('domain') or not user_params.has_key('configid'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'DeleteSpecificStagingConfig', 'DomainName':user_params['domain'], 'ConfigId':user_params['configid']}
   
       elif user_params['action'] == 'del_product_env':
           if not user_params.has_key('domain') or not user_params.has_key('configid'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'DeleteSpecificConfig', 'DomainName':user_params['domain'], 'ConfigId':user_params['configid']}
   
       elif user_params['action'] == 'publish_test_env':
           if not user_params.has_key('domain'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'PublishStagingConfigToProduction', 'DomainName':user_params['domain'], 'FunctionName':'edge_function'}
   
       elif user_params['action'] == 'rollback_test_env':
           if not user_params.has_key('domain'):
               parser.print_help()
               sys.exit(0)
           req_args = {'Action':'RollbackStagingConfig', 'DomainName':user_params['domain'], 'FunctionName':'edge_function'}
   
       else:
           parser.print_help()
           sys.exit(0)
   
       return req_args
   
   if __name__ == '__main__':
       parser = OptionParser("%s Action=action Param1=Value1 Param2=Value2 %s\n" % (sys.argv[0], cmdlist))
       parser.add_option("-i", "--id", dest="accesskeyid", help="specify access key id")
       parser.add_option("-s", "--secret", dest="accesskeysecret", help="specify access key secret")
   
       (options, args) = parser.parse_args()
       if len(args) < 1:
           parser.print_help()
           sys.exit(0)
   
       if args[0] == 'help':
           parser.print_help()
           sys.exit(0)
       if args[0] != 'config':
           setup_credentials()
       else: #it's a configure id/secret command
           configure_accesskeypair(args, options)
           sys.exit(0)
   
       user_params = {}
       idx = 1
       if sys.argv[1].lower().startswith('action='):
           _, value = sys.argv[1].split('=')
           user_params['action'] = value
           idx = 2
       else:
           parser.print_help()
           sys.exit(0)
   
       for arg in sys.argv[idx:]:
           try:
               key, value = arg.split('=', 1)
               if key == 'rule': # push_test_env / push_product_env
                   if not user_params.has_key('rule'):
                       user_params['rule'] = []
                   user_params['rule'].append(json.loads(value))
               else:
                   user_params[key.strip()] = value
           except ValueError, e:
               print(e.read().strip())
               raise SystemExit(e)
   
       req_args = parse_args(user_params)
       make_request(req_args)

Configure credentials

Run the following command to store your AccessKey pair in aliyun.ini:

python ./es.py config --id=<your-access-key-id> --secret=<your-access-key-secret>

Verify the credentials were saved:

cat aliyun.ini

Expected output:

[Credentials]
accesskeyid = <your-access-key-id>
accesskeysecret = <your-access-key-secret>

Publish a script

Push a script to the staging environment to test it before going live:

./es.py action=push_test_env domain=<domain> rule='{"pos":"head","pri":"10","rule_path":"./m3u8.es","enable":"on"}'

To push a script directly to the production environment:

./es.py action=push_product_env domain=<domain> rule='{"pos":"head","pri":"10","rule_path":"./m3u8.es","enable":"on","configid":"12345"}'

The rule parameter accepts the following fields:

You can specify multiple rule parameters in a single command to push multiple scripts at once.

About configuration IDs

A configuration ID (configid) is generated automatically when you create a script for a domain name. You do not need to provide one when creating a script. To modify an existing script, query it first to get its configuration ID, then include "configid":"<configid>" in the rule field. For details, see Usage notes on ConfigId.

Query scripts

List all EdgeScript scripts for a domain name:

# Staging environment
./es.py action=query_test_env domain=<domain>

# Production environment
./es.py action=query_product_env domain=<domain>

Delete a script

Delete a script by its configuration ID. Run a query first if you don't know the configuration ID.

# Staging environment
./es.py action=del_test_env domain=<domain> configid=<configid>

# Production environment
./es.py action=del_product_env domain=<domain> configid=<configid>

Promote staging scripts to production

After validating scripts in the staging environment, promote them to production:

./es.py action=publish_test_env domain=<domain>

Roll back scripts

Roll back the staging environment to its previous configuration:

./es.py action=rollback_test_env domain=<domain>