mirror of
https://github.com/GAM-team/GAM.git
synced 2026-06-17 04:31:37 +00:00
Upgrade yubkey for 5.0 release. Fixes #1587
This commit is contained in:
@@ -1314,34 +1314,38 @@ def doCheckServiceAccount(users):
|
||||
3,
|
||||
'Invalid private key in oauth2service.json. Please delete the file and then\nrecreate with "gam create project" or "gam use project"'
|
||||
)
|
||||
print(
|
||||
'Checking key age. Google recommends rotating keys on a routine basis...'
|
||||
key_type = GM_Globals[GM_OAUTH2SERVICE_JSON_DATA].get('key_type', 'default')
|
||||
if key_type == 'yubikey':
|
||||
printPassFail('Skipping age check. YubiKey rotation not necessary.', test_pass)
|
||||
else:
|
||||
print(
|
||||
'Checking key age. Google recommends rotating keys on a routine basis...'
|
||||
)
|
||||
try:
|
||||
iam = buildGAPIServiceObject('iam', None)
|
||||
project = GM_Globals[GM_OAUTH2SERVICE_ACCOUNT_CLIENT_ID]
|
||||
key_id = GM_Globals[GM_OAUTH2SERVICE_JSON_DATA]['private_key_id']
|
||||
name = f'projects/-/serviceAccounts/{project}/keys/{key_id}'
|
||||
key = gapi.call(iam.projects().serviceAccounts().keys(),
|
||||
'get',
|
||||
name=name,
|
||||
throw_reasons=[gapi_errors.ErrorReason.FOUR_O_THREE])
|
||||
key_created = dateutil.parser.parse(
|
||||
key['validAfterTime'], ignoretz=True)
|
||||
key_age = datetime.datetime.now() - key_created
|
||||
key_days = key_age.days
|
||||
if key_days > 30:
|
||||
print(
|
||||
'Your key is old. Recommend running "gam rotate sakey" to get a new key'
|
||||
)
|
||||
try:
|
||||
iam = buildGAPIServiceObject('iam', None)
|
||||
project = GM_Globals[GM_OAUTH2SERVICE_ACCOUNT_CLIENT_ID]
|
||||
key_id = GM_Globals[GM_OAUTH2SERVICE_JSON_DATA]['private_key_id']
|
||||
name = f'projects/-/serviceAccounts/{project}/keys/{key_id}'
|
||||
key = gapi.call(iam.projects().serviceAccounts().keys(),
|
||||
'get',
|
||||
name=name,
|
||||
throw_reasons=[gapi_errors.ErrorReason.FOUR_O_THREE])
|
||||
key_created = dateutil.parser.parse(
|
||||
key['validAfterTime'], ignoretz=True)
|
||||
key_age = datetime.datetime.now() - key_created
|
||||
key_days = key_age.days
|
||||
if key_days > 30:
|
||||
print(
|
||||
'Your key is old. Recommend running "gam rotate sakey" to get a new key'
|
||||
)
|
||||
key_age_result = test_warn
|
||||
else:
|
||||
key_age_result = test_pass
|
||||
except googleapiclient.errors.HttpError:
|
||||
key_age_result = test_warn
|
||||
else:
|
||||
key_age_result = test_pass
|
||||
except googleapiclient.errors.HttpError:
|
||||
key_age_result = test_warn
|
||||
key_days = 'UNKNOWN'
|
||||
print('Unable to check key age, please run "gam update project"')
|
||||
printPassFail(f'Key is {key_days} days old', key_age_result)
|
||||
key_days = 'UNKNOWN'
|
||||
print('Unable to check key age, please run "gam update project"')
|
||||
printPassFail(f'Key is {key_days} days old', key_age_result)
|
||||
if not check_scopes:
|
||||
for _, scopes in list(API_SCOPE_MAPPING.items()):
|
||||
for scope in scopes:
|
||||
@@ -4473,15 +4477,7 @@ def getImap(users):
|
||||
soft_errors=True,
|
||||
userId='me')
|
||||
if result:
|
||||
enabled = result['enabled']
|
||||
if enabled:
|
||||
print(
|
||||
f'User: {user}, IMAP Enabled: {enabled}, autoExpunge: {result["autoExpunge"]}, expungeBehavior: {result["expungeBehavior"]}, maxFolderSize: {result["maxFolderSize"]}{currentCount(i, count)}'
|
||||
)
|
||||
else:
|
||||
print(
|
||||
f'User: {user}, IMAP Enabled: {enabled}{currentCount(i, count)}'
|
||||
)
|
||||
display.print_json(result)
|
||||
|
||||
|
||||
def doPop(users):
|
||||
@@ -7912,6 +7908,7 @@ def doCreateOrRotateServiceAccountKeys(iam=None,
|
||||
yk = yubikey.YubiKey(new_data)
|
||||
if 'yubikey_serial_number' not in new_data:
|
||||
new_data['yubikey_serial_number'] = yk.get_serial_number()
|
||||
yk = yubikey.YubiKey(new_data)
|
||||
if 'yubikey_slot' not in new_data:
|
||||
new_data['yubikey_slot'] = 'AUTHENTICATION'
|
||||
publicKeyData = yk.get_certificate()
|
||||
@@ -12069,6 +12066,7 @@ def ProcessGAMCommand(args):
|
||||
action = sys.argv[2].lower().replace('_', '')
|
||||
if action == 'resetpiv':
|
||||
yk = yubikey.YubiKey()
|
||||
yk.serial_number = yk.get_serial_number()
|
||||
yk.reset_piv()
|
||||
sys.exit(0)
|
||||
users = getUsersToModify()
|
||||
|
||||
@@ -8,7 +8,7 @@ from threading import Timer
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import padding
|
||||
from smartcard.Exceptions import CardConnectionException
|
||||
from ykman.device import connect_to_device
|
||||
from ykman.device import list_all_devices
|
||||
from ykman.piv import generate_self_signed_certificate, \
|
||||
generate_chuid
|
||||
from yubikit.piv import DEFAULT_MANAGEMENT_KEY, \
|
||||
@@ -20,11 +20,14 @@ from yubikit.piv import DEFAULT_MANAGEMENT_KEY, \
|
||||
OBJECT_ID, \
|
||||
SLOT, \
|
||||
TOUCH_POLICY
|
||||
from yubikit.core.smartcard import ApduError
|
||||
from yubikit.core.smartcard import ApduError, \
|
||||
SmartCardConnection
|
||||
from gam import controlflow
|
||||
|
||||
|
||||
class YubiKey():
|
||||
|
||||
|
||||
def __init__(self, service_account_info=None):
|
||||
self.key_type = None
|
||||
self.slot = None
|
||||
@@ -46,12 +49,16 @@ class YubiKey():
|
||||
self.pin = service_account_info.get('yubikey_pin')
|
||||
self.key_id = service_account_info.get('private_key_id')
|
||||
|
||||
|
||||
def _connect(self):
|
||||
try:
|
||||
conn, _, _ = connect_to_device(self.serial_number)
|
||||
devices = list_all_devices()
|
||||
for (device, info) in devices:
|
||||
if info.serial == self.serial_number:
|
||||
return device.open_connection(SmartCardConnection)
|
||||
except CardConnectionException as err:
|
||||
controlflow.system_error_exit(9, f'YubiKey - {err}')
|
||||
return conn
|
||||
|
||||
|
||||
def get_certificate(self):
|
||||
try:
|
||||
@@ -79,11 +86,22 @@ class YubiKey():
|
||||
|
||||
def get_serial_number(self):
|
||||
try:
|
||||
_, _, info = connect_to_device(self.serial_number)
|
||||
return info.serial
|
||||
devices = list_all_devices()
|
||||
if self.serial_number:
|
||||
for (device, info) in devices:
|
||||
if info.serial == self.serial_number:
|
||||
return info.serial
|
||||
msg = f'Could not find YubiKey with serial {self.serial_number}'
|
||||
controlflow.system_error_exit(3, msg)
|
||||
if len(devices) > 1:
|
||||
serials = ', '.join([str(info.serial) for (_, info) in devices])
|
||||
msg = f'Multiple YubiKeys connected. Specify yubikey_serial_number and one of {serials}'
|
||||
controlflow.system_error_exit(4, msg)
|
||||
return devices[0][1].serial
|
||||
except ValueError as err:
|
||||
controlflow.system_error_exit(9, f'YubiKey - {err}')
|
||||
|
||||
|
||||
def reset_piv(self):
|
||||
'''Resets YubiKey PIV app and generates new key for GAM to use.'''
|
||||
reply = str(input('This will wipe all PIV keys and configuration from your YubiKey. Are you sure? (y/N) ').lower().strip())
|
||||
@@ -95,7 +113,9 @@ class YubiKey():
|
||||
piv = PivSession(conn)
|
||||
piv.reset()
|
||||
rnd = SystemRandom()
|
||||
pin_puk_chars = string.ascii_letters + string.digits + string.punctuation
|
||||
pin_puk_chars = string.ascii_letters + \
|
||||
string.digits + \
|
||||
string.punctuation
|
||||
new_puk = ''.join(rnd.choice(pin_puk_chars) for _ in range(8))
|
||||
new_pin = ''.join(rnd.choice(pin_puk_chars) for _ in range(8))
|
||||
piv.change_puk('12345678', new_puk)
|
||||
@@ -155,3 +175,4 @@ class YubiKey():
|
||||
if 'mplock' in globals():
|
||||
mplock.release()
|
||||
return signed
|
||||
|
||||
|
||||
@@ -10,4 +10,4 @@ importlib.metadata; python_version < '3.8'
|
||||
passlib>=1.7.2
|
||||
pathvalidate
|
||||
python-dateutil
|
||||
yubikey-manager>=4.0.0
|
||||
yubikey-manager>=5.0
|
||||
|
||||
Reference in New Issue
Block a user