Migrate Origin Trials token generator scripts to Python 3
Bug: 1274995
Change-Id: I6a67c82bb2b6c23275001d3e3e9d5f3cdf15ec44
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/3338894
Reviewed-by: Ian Clelland <iclelland@chromium.org>
Commit-Queue: Daniel Smith <danielrsmith@google.com>
Cr-Commit-Position: refs/heads/main@{#959445}
This commit is contained in:
Daniel Smith
committed by
Chromium LUCI CQ
Chromium LUCI CQ
parent
6354562dab
commit
b4f30fb7e2
tools/origin_trials
@ -2,21 +2,19 @@
|
||||
# Use of this source code is governed by a BSD-style license that can be
|
||||
# found in the LICENSE file.
|
||||
|
||||
USE_PYTHON3 = False
|
||||
USE_PYTHON3 = True
|
||||
|
||||
|
||||
def _CommonChecks(input_api, output_api):
|
||||
results = []
|
||||
|
||||
# Run Pylint over the files in the directory.
|
||||
# TODO(crbug.com/1262279): Enable these warnings after migrating to Python3.
|
||||
disabled_warnings = ('super-with-arguments', )
|
||||
pylint_checks = input_api.canned_checks.GetPylint(
|
||||
input_api, output_api, disabled_warnings=disabled_warnings, version='2.7')
|
||||
pylint_checks = input_api.canned_checks.GetPylint(input_api,
|
||||
output_api,
|
||||
version='2.7')
|
||||
results.extend(input_api.RunTests(pylint_checks))
|
||||
|
||||
# Run the generate_token unittests.
|
||||
#TODO(https://crbug.com/1274995): Run the tests on Python3.
|
||||
results.extend(
|
||||
input_api.canned_checks.RunUnitTestsInDirectory(
|
||||
input_api,
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
#!/usr/bin/env python
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) 2017 The Chromium Authors. All rights reserved.
|
||||
# Use of this source code is governed by a BSD-style license that can be
|
||||
# found in the LICENSE file.
|
||||
@ -43,28 +43,54 @@ PAYLOAD_LENGTH_SIZE = 4
|
||||
PAYLOAD_OFFSET = PAYLOAD_LENGTH_OFFSET + PAYLOAD_LENGTH_SIZE
|
||||
|
||||
# This script supports Version 2 and Version 3 tokens.
|
||||
VERSION2 = "\x02"
|
||||
VERSION3 = "\x03"
|
||||
VERSION2 = b'\x02'
|
||||
VERSION3 = b'\x03'
|
||||
|
||||
# Only empty string and "subset" are supported in alternative usage restriction.
|
||||
USAGE_RESTRICTION = ["", "subset"]
|
||||
|
||||
# Chrome public key, used by default to validate signatures
|
||||
# - Copied from chrome/common/origin_trials/chrome_origin_trial_policy.cc
|
||||
CHROME_PUBLIC_KEY = [
|
||||
0x7c, 0xc4, 0xb8, 0x9a, 0x93, 0xba, 0x6e, 0xe2, 0xd0, 0xfd, 0x03,
|
||||
0x1d, 0xfb, 0x32, 0x66, 0xc7, 0x3b, 0x72, 0xfd, 0x54, 0x3a, 0x07,
|
||||
0x51, 0x14, 0x66, 0xaa, 0x02, 0x53, 0x4e, 0x33, 0xa1, 0x15,
|
||||
]
|
||||
CHROME_PUBLIC_KEY = bytes([
|
||||
0x7c,
|
||||
0xc4,
|
||||
0xb8,
|
||||
0x9a,
|
||||
0x93,
|
||||
0xba,
|
||||
0x6e,
|
||||
0xe2,
|
||||
0xd0,
|
||||
0xfd,
|
||||
0x03,
|
||||
0x1d,
|
||||
0xfb,
|
||||
0x32,
|
||||
0x66,
|
||||
0xc7,
|
||||
0x3b,
|
||||
0x72,
|
||||
0xfd,
|
||||
0x54,
|
||||
0x3a,
|
||||
0x07,
|
||||
0x51,
|
||||
0x14,
|
||||
0x66,
|
||||
0xaa,
|
||||
0x02,
|
||||
0x53,
|
||||
0x4e,
|
||||
0x33,
|
||||
0xa1,
|
||||
0x15,
|
||||
])
|
||||
|
||||
# Default key file, relative to script_dir.
|
||||
DEFAULT_KEY_FILE = 'eftest.key'
|
||||
|
||||
|
||||
class OverrideKeyFileAction(argparse.Action):
|
||||
def __init__(self, option_strings, dest, **kwargs):
|
||||
super(OverrideKeyFileAction, self).__init__(option_strings, dest, **kwargs)
|
||||
|
||||
def __call__(self, parser, namespace, values, option_string=None):
|
||||
setattr(namespace, "use_chrome_key", None)
|
||||
setattr(namespace, self.dest, values)
|
||||
@ -101,7 +127,7 @@ def main():
|
||||
private_key_file = args.key_file
|
||||
else:
|
||||
if (args.use_chrome_key):
|
||||
public_key = "".join(chr(x) for x in CHROME_PUBLIC_KEY)
|
||||
public_key = CHROME_PUBLIC_KEY
|
||||
else:
|
||||
# Use the test key, relative to this script.
|
||||
private_key_file = os.path.join(script_dir, DEFAULT_KEY_FILE)
|
||||
@ -145,7 +171,7 @@ def main():
|
||||
version_number = 0
|
||||
for x in version:
|
||||
version_number <<= 8
|
||||
version_number += ord(x)
|
||||
version_number += x
|
||||
if (version not in (VERSION2, VERSION3)):
|
||||
print("Token has wrong version: %d" % version_number)
|
||||
sys.exit(1)
|
||||
@ -237,8 +263,9 @@ def main():
|
||||
print(" Usage Restriction: %s" % usage_restriction)
|
||||
print(" Feature: %s" % trial_name)
|
||||
print(" Expiry: %d (%s UTC)" % (expiry, datetime.utcfromtimestamp(expiry)))
|
||||
print(" Signature: %s" % ", ".join('0x%02x' % ord(x) for x in signature))
|
||||
print(" Signature (Base64): %s" % base64.b64encode(signature))
|
||||
print(" Signature: %s" % ", ".join('0x%02x' % x for x in signature))
|
||||
b64_signature = base64.b64encode(signature).decode("ascii")
|
||||
print(" Signature (Base64): %s" % b64_signature)
|
||||
print()
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
#!/usr/bin/env python
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) 2016 The Chromium Authors. All rights reserved.
|
||||
# Use of this source code is governed by a BSD-style license that can be
|
||||
# found in the LICENSE file.
|
||||
@ -30,12 +30,7 @@ import time
|
||||
from datetime import datetime
|
||||
|
||||
from six import raise_from
|
||||
|
||||
try:
|
||||
from urllib.parse import urlparse
|
||||
except ImportError:
|
||||
# ToDo: Remove Exception case upon full migration to Python 3
|
||||
from urlparse import urlparse
|
||||
from urllib.parse import urlparse
|
||||
|
||||
script_dir = os.path.dirname(os.path.realpath(__file__))
|
||||
sys.path.insert(0, os.path.join(script_dir, 'third_party', 'ed25519'))
|
||||
@ -45,8 +40,8 @@ import ed25519
|
||||
# no longer than 63 ASCII characters)
|
||||
DNS_LABEL_REGEX = re.compile(r"^(?!-)[a-z\d-]{1,63}(?<!-)$", re.IGNORECASE)
|
||||
|
||||
# This script generates Version 2 and 3 tokens.
|
||||
VERSION = {"2": (2, "\x02"), "3": (3, "\x03")}
|
||||
# Only Version 2 and Version 3 are currently supported.
|
||||
VERSIONS = {"2": (2, b'\x02'), "3": (3, b'\x03')}
|
||||
|
||||
# Only empty string and "subset" are currently supoprted in alternative usage
|
||||
# resetriction.
|
||||
@ -60,12 +55,10 @@ def VersionFromArg(arg):
|
||||
"""Determines whether a string represents a valid version.
|
||||
Only Version 2 and Version 3 are currently supported.
|
||||
|
||||
Returns a tuple (version number, version byte) if version is valid.
|
||||
Returns a tuple of the int and bytes representation of version.
|
||||
Returns None if version is not valid.
|
||||
"""
|
||||
if not arg or len(arg) > 1:
|
||||
return None
|
||||
return VERSION.get(arg, None)
|
||||
return VERSIONS.get(arg, None)
|
||||
|
||||
|
||||
def HostnameFromArg(arg):
|
||||
@ -136,12 +129,14 @@ def GenerateTokenData(version, origin, is_subdomain, is_third_party,
|
||||
def GenerateDataToSign(version, data):
|
||||
return version + struct.pack(">I",len(data)) + data
|
||||
|
||||
|
||||
def Sign(private_key, data):
|
||||
return ed25519.signature(data, private_key[:32], private_key[32:])
|
||||
|
||||
|
||||
def FormatToken(version, signature, data):
|
||||
return base64.b64encode(version + signature +
|
||||
struct.pack(">I",len(data)) + data)
|
||||
return base64.b64encode(version + signature + struct.pack(">I", len(data)) +
|
||||
data).decode("ascii")
|
||||
|
||||
|
||||
def ParseArgs():
|
||||
@ -217,8 +212,10 @@ def GenerateTokenAndSignature():
|
||||
args = ParseArgs()
|
||||
expiry = ExpiryFromArgs(args)
|
||||
|
||||
key_file = open(os.path.expanduser(args.key_file), mode="rb")
|
||||
private_key = key_file.read(64)
|
||||
version_int, version_bytes = args.version
|
||||
|
||||
with open(os.path.expanduser(args.key_file), mode="rb") as key_file:
|
||||
private_key = key_file.read(64)
|
||||
|
||||
# Validate that the key file read was a proper Ed25519 key -- running the
|
||||
# publickey method on the first half of the key should return the second
|
||||
@ -228,16 +225,16 @@ def GenerateTokenAndSignature():
|
||||
print("Unable to use the specified private key file.")
|
||||
sys.exit(1)
|
||||
|
||||
if (not args.version):
|
||||
if (not version_int):
|
||||
print("Invalid token version. Only version 2 and 3 are supported.")
|
||||
sys.exit(1)
|
||||
|
||||
if (args.is_third_party is not None and args.version[0] != 3):
|
||||
if (args.is_third_party is not None and version_int != 3):
|
||||
print("Only version 3 token supports is_third_party flag.")
|
||||
sys.exit(1)
|
||||
|
||||
if (args.usage_restriction is not None):
|
||||
if (args.version[0] != 3):
|
||||
if (version_int != 3):
|
||||
print("Only version 3 token supports alternative usage restriction.")
|
||||
sys.exit(1)
|
||||
if (args.usage_restriction not in USAGE_RESTRICTION):
|
||||
@ -245,12 +242,10 @@ def GenerateTokenAndSignature():
|
||||
"Only empty string and \"subset\" are supported in alternative usage "
|
||||
"restriction.")
|
||||
sys.exit(1)
|
||||
|
||||
token_data = GenerateTokenData(args.version[0], args.origin,
|
||||
args.is_subdomain, args.is_third_party,
|
||||
args.usage_restriction, args.trial_name,
|
||||
expiry)
|
||||
data_to_sign = GenerateDataToSign(args.version[1], token_data)
|
||||
token_data = GenerateTokenData(version_int, args.origin, args.is_subdomain,
|
||||
args.is_third_party, args.usage_restriction,
|
||||
args.trial_name, expiry)
|
||||
data_to_sign = GenerateDataToSign(version_bytes, token_data)
|
||||
signature = Sign(private_key, data_to_sign)
|
||||
|
||||
# Verify that that the signature is correct before printing it.
|
||||
@ -261,34 +256,35 @@ def GenerateTokenAndSignature():
|
||||
print("(The original error was: %s)" % exc)
|
||||
sys.exit(1)
|
||||
|
||||
token_data = GenerateTokenData(args.version[0], args.origin,
|
||||
args.is_subdomain, args.is_third_party,
|
||||
args.usage_restriction, args.trial_name,
|
||||
expiry)
|
||||
data_to_sign = GenerateDataToSign(args.version[1], token_data)
|
||||
token_data = GenerateTokenData(version_int, args.origin, args.is_subdomain,
|
||||
args.is_third_party, args.usage_restriction,
|
||||
args.trial_name, expiry)
|
||||
data_to_sign = GenerateDataToSign(version_bytes, token_data)
|
||||
signature = Sign(private_key, data_to_sign)
|
||||
return args, token_data, signature, expiry
|
||||
|
||||
|
||||
def main():
|
||||
args, token_data, signature, expiry = GenerateTokenAndSignature()
|
||||
version_int, version_bytes = args.version
|
||||
|
||||
# Output the token details
|
||||
print("Token details:")
|
||||
print(" Version: %s" % args.version[0])
|
||||
print(" Version: %s" % version_int)
|
||||
print(" Origin: %s" % args.origin)
|
||||
print(" Is Subdomain: %s" % args.is_subdomain)
|
||||
if args.version[0] == 3:
|
||||
if version_int == 3:
|
||||
print(" Is Third Party: %s" % args.is_third_party)
|
||||
print(" Usage Restriction: %s" % args.usage_restriction)
|
||||
print(" Feature: %s" % args.trial_name)
|
||||
print(" Expiry: %d (%s UTC)" % (expiry, datetime.utcfromtimestamp(expiry)))
|
||||
print(" Signature: %s" % ", ".join('0x%02x' % ord(x) for x in signature))
|
||||
print(" Signature (Base64): %s" % base64.b64encode(signature))
|
||||
print(" Signature: %s" % ", ".join('0x%02x' % x for x in signature))
|
||||
b64_signature = base64.b64encode(signature).decode("ascii")
|
||||
print(" Signature (Base64): %s" % b64_signature)
|
||||
print()
|
||||
|
||||
# Output the properly-formatted token.
|
||||
print(FormatToken(args.version[1], signature, token_data))
|
||||
print(FormatToken(version_bytes, signature, token_data))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
#!/usr/bin/env python
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) 2016 The Chromium Authors. All rights reserved.
|
||||
# Use of this source code is governed by a BSD-style license that can be
|
||||
# found in the LICENSE file.
|
||||
@ -7,12 +7,10 @@
|
||||
|
||||
import argparse
|
||||
import generate_token
|
||||
import unittest
|
||||
# TODO(https://crbug.com/1274995): Use unittest.mock after migrating to Python3.
|
||||
from mock import patch as mock_patch
|
||||
from unittest import main, mock, TestCase
|
||||
|
||||
|
||||
class GenerateTokenTest(unittest.TestCase):
|
||||
class GenerateTokenTest(TestCase):
|
||||
|
||||
def test_hostname_validation(self):
|
||||
for hostname, expected_result in [
|
||||
@ -67,10 +65,22 @@ class GenerateTokenTest(unittest.TestCase):
|
||||
invalid_hostname)
|
||||
|
||||
def test_end_to_end(self):
|
||||
with mock_patch('sys.argv',
|
||||
with mock.patch('sys.argv',
|
||||
['generate-token.py', 'example.com', 'example']):
|
||||
generate_token.GenerateTokenAndSignature()
|
||||
|
||||
def test_FormatToken(self):
|
||||
for version, signature, token_data, expected in [
|
||||
(b'\x03', bytes([1, 2, 3]), bytes([4, 5, 6]), 'AwECAwAAAAMEBQY='),
|
||||
(b'\x03', bytes([200, 100, 1]), bytes([30, 40,
|
||||
50]), 'A8hkAQAAAAMeKDI='),
|
||||
(b'\x02', bytes([2, 3, 2]), bytes([2, 3, 2]), 'AgIDAgAAAAMCAwI='),
|
||||
(b'\x02', bytes([255, 150, 10]), bytes([10, 150,
|
||||
255]), 'Av+WCgAAAAMKlv8=')
|
||||
]:
|
||||
self.assertEqual(
|
||||
generate_token.FormatToken(version, signature, token_data), expected)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
main()
|
||||
|
||||
@ -54,22 +54,20 @@ def scalarmult(P,e):
|
||||
|
||||
def encodeint(y):
|
||||
bits = [(y >> i) & 1 for i in range(b)]
|
||||
return ''.join([
|
||||
chr(sum([bits[i * 8 + j] << j for j in range(8)])) for i in range(b // 8)
|
||||
])
|
||||
return bytes(
|
||||
[sum([bits[i * 8 + j] << j for j in range(8)]) for i in range(b // 8)])
|
||||
|
||||
|
||||
def encodepoint(P):
|
||||
x = P[0]
|
||||
y = P[1]
|
||||
bits = [(y >> i) & 1 for i in range(b - 1)] + [x & 1]
|
||||
return ''.join([
|
||||
chr(sum([bits[i * 8 + j] << j for j in range(8)])) for i in range(b // 8)
|
||||
])
|
||||
return bytes(
|
||||
[sum([bits[i * 8 + j] << j for j in range(8)]) for i in range(b // 8)])
|
||||
|
||||
|
||||
def bit(h,i):
|
||||
return (ord(h[i // 8]) >> (i % 8)) & 1
|
||||
return (h[i // 8] >> (i % 8)) & 1
|
||||
|
||||
|
||||
def publickey(sk):
|
||||
@ -85,7 +83,7 @@ def Hint(m):
|
||||
def signature(m,sk,pk):
|
||||
h = H(sk)
|
||||
a = 2**(b-2) + sum(2**i * bit(h,i) for i in range(3,b-2))
|
||||
r = Hint(''.join([h[i] for i in range(b // 8, b // 4)]) + m)
|
||||
r = Hint(bytes([h[i] for i in range(b // 8, b // 4)]) + m)
|
||||
R = scalarmult(B,r)
|
||||
S = (r + Hint(encodepoint(R) + pk + m) * a) % l
|
||||
return encodepoint(R) + encodeint(S)
|
||||
|
||||
Reference in New Issue
Block a user