UTV验证

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

137

138

139

140

141

142

143

144

145

146

147

148

149

#!/usr/bin/env python3

# coding: utf-8

import sys

import traceback

import argparse

import time

import hashlib

import hmac

import base64

import re

import pyDes

from urllib.parse import urlparse, parse_qsl, urlencode, unquote_plus

parser = argparse.ArgumentParser()

parser.add_argument("--method", choices = ["md5", "hmac", "3des", "de3des"], default = "md5", help = "Method for UTV")

parser.add_argument("--url", required = True, help = "The URL which needs to apply UTV")

parser.add_argument("--secret", required = True, help = "The encrypt key for UTV")

parser.add_argument("--hex", action = "store_true", default = False, help = "Time in HEX format")

parser.add_argument("--expire", metavar = "seconds", type = int, default = 0, help = "Expire seconds add to time allow")

parser.add_argument("--basetime", metavar = "UNIX_time_seconds", type = float, default = time.time(), help = "Base time (For development confirm)")

parser.add_argument("--miaopai", action = "store_true", default = False, help = "Generate UTV for Miaopai")

parser.add_argument("--secret_3des", help = "The encrypt key for 3DES hash")

def get_md5_url(base_url, secret, base_time, hex, expire, miaopai = False):

    url_parts = urlparse(base_url)

    url_host = url_parts.netloc

    url_path = url_parts.path

    url_query = url_parts.query

    if miaopai:

        expire = base_time * 1000 + expire * 1000

        expire = int(expire)

    else:

        expire = base_time + expire

        expire = int(expire)

    if hex:

        expire = str(format(expire, ‘x‘))

    else:

        expire = str(expire)

    if miaopai:

        match = re.match(r‘^/stream/(.*)\.(.*)($|\?.*)‘, url_path)

        if match:

            url_data = match.group(1)

            print(url_data)

        str_byte = (url_data + secret + expire).encode(‘utf-8‘)

    else:

        str_byte = (secret + url_path + expire).encode(‘utf-8‘)

    m = hashlib.md5()

    m.update(str_byte)

    hmac = m.hexdigest()

    url_query_params = parse_qsl(url_query, keep_blank_values = True)

    if miaopai:

        url_query_params.append((‘time_stamp‘, expire))

        url_query_params.append((‘ssig‘, hmac))

    else:

        url_query_params.append((‘px-time‘, expire))

        url_query_params.append((‘px-hash‘, hmac))

    url_query_param = urlencode(url_query_params)

    re_url = url_parts.scheme + "://" + url_host + url_path + "?" + url_query_param

    re_url = re_url.replace(‘=&‘, ‘&‘)

    re_url = re.sub(r‘=$‘, ‘‘, re_url)

    return re_url

def get_hmac_url(base_url, secret, base_time, expire):

    url_parts = urlparse(base_url)

    url_host = url_parts.netloc

    url_path = url_parts.path

    url_query = url_parts.query

    expire = base_time + expire

    str_sign = (‘GET‘ + ‘\n‘ + ‘\n‘ + ‘\n‘ + str(expire) + ‘\n‘ + ‘/‘ + url_host + url_path).encode(‘utf-8‘)

    str_ssig = base64.b64encode(hmac.new(bytearray(secret, ‘utf-8‘), str_sign, hashlib.sha1).digest())[5:15]

    url_query_params = parse_qsl(url_query, keep_blank_values = True)

    url_query_params.append((‘ssig‘, str_ssig))

    url_query_params.append((‘Expires‘, expire))

    url_query_param = urlencode(url_query_params)

    return url_parts.scheme + ‘://‘ + url_host + url_path + ‘?‘ + url_query_param

def get_3des_url(base_url, secret_3des, secret, base_time, hex, expire):

    url_parts = urlparse(base_url)

    url_host = url_parts.netloc

    url_path = url_parts.path

    url_query = url_parts.query

    calc_path = url_path[1:]

    calc_byte = calc_path.encode(‘utf-8‘)

    secret_byte = bytearray(secret_3des, ‘utf-8‘)

    secret_byte_24 = bytearray(‘‘, ‘utf-8‘)

    if len(secret_byte) < 24:

        for x in range(0, 24 - len(secret_byte)):

            secret_byte.append(0b0)

    elif len(secret_byte) > 24:

        x = 0

        while(x < 24):

            secret_byte_24.append(secret_byte[x])

            x += 1

        secret_byte = secret_byte_24

    k = pyDes.triple_des(secret_byte, pyDes.ECB, b"\0\0\0\0\0\0\0\0", pad = None, padmode = pyDes.PAD_PKCS5)

    d = k.encrypt(calc_byte)

    url_query_params = parse_qsl(url_query, keep_blank_values = True)

    url_query_params.append((‘encrypt‘, str(1)))

    url_query_param = urlencode(url_query_params)

    url_path = unquote_plus(str(base64.b64encode(d)), encoding=‘utf-8‘).replace(‘ ‘, ‘-‘)[2:-1]

    des3_url = url_parts.scheme + ‘://‘ + url_host + ‘/‘ + url_path + ‘?‘ + url_query_param

    return(get_md5_url(des3_url, secret, base_time, hex, expire))

def decode_3des_url(base_url, secret_3des):

    url_parts = urlparse(base_url)

    url_host = url_parts.netloc

    url_path = url_parts.path

    url_query = url_parts.query

    de_url_path = url_path.replace(‘-‘, ‘+‘)

    de_url_path = base64.b64decode(de_url_path[1:].encode(‘utf-8‘))

    secret_byte = bytearray(secret_3des, ‘utf-8‘)

    secret_byte_24 = bytearray(‘‘, ‘utf-8‘)

    if len(secret_byte) < 24:

        for x in range(0, 24 - len(secret_byte)):

            secret_byte.append(0b0)

    elif len(secret_byte) > 24:

        x = 0

        while(x < 24):

            secret_byte_24.append(secret_byte[x])

            x += 1

        secret_byte = secret_byte_24

    k = pyDes.triple_des(secret_byte, pyDes.ECB, b"\0\0\0\0\0\0\0\0", pad = None, padmode = pyDes.PAD_PKCS5)

    d = k.decrypt(de_url_path)

    url_path = str(d)[2:-1]

    return url_parts.scheme + ‘://‘ + url_host + ‘/‘ + url_path

def main():

    args = parser.parse_args()

    utv_method = args.method

    utv_url = args.url

    utv_secret = args.secret

    utv_hex = args.hex

    utv_expire = args.expire

    utv_basetime = args.basetime

    utv_miaopai = args.miaopai

    utv_secret_3des = args.secret_3des

    if utv_method == "md5":

        print(get_md5_url(utv_url, utv_secret, utv_basetime, utv_hex, utv_expire, utv_miaopai))

    elif utv_method == "hmac":

        print(get_hmac_url(utv_url, utv_secret, utv_basetime, utv_expire))

    elif utv_method == "3des":

        print(get_3des_url(utv_url, utv_secret_3des, utv_secret, utv_basetime, utv_hex, utv_expire))

    elif utv_method == "de3des":

        print(decode_3des_url(utv_url, utv_secret))

    exit()

if __name__ == ‘__main__‘:

    main()