kubernetes-client/python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #79 from bpicolo/fix_base64_padding_for_kconfig

Browse Source 
Fix base64 padding for kube config
This commit is contained in:
Haowei Cai (Roy) 2019-06-19 15:46:02 -07:00 committed by GitHub
commit 487c9182c8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 114 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
config/kube_config.py
View File

@ -277,18 +277,31 @@ class KubeConfigLoader(object):
if 'config' not in provider:
return
parts = provider['config']['id-token'].split('.')
reserved_characters = frozenset(["=", "+", "/"])
token = provider['config']['id-token']
if any(char in token for char in reserved_characters):
# Invalid jwt, as it contains url-unsafe chars
return
parts = token.split('.')
if len(parts) != 3: # Not a valid JWT
return None
return
padding = (4 - len(parts[1]) % 4) * '='
if len(padding) == 3:
# According to spec, 3 padding characters cannot occur
# in a valid jwt
# https://tools.ietf.org/html/rfc7515#appendix-C
return
if PY3:
jwt_attributes = json.loads(
base64.b64decode(parts[1]).decode('utf-8')
base64.b64decode(parts[1] + padding).decode('utf-8')
)
else:
jwt_attributes = json.loads(
base64.b64decode(parts[1] + "==")
base64.b64decode(parts[1] + padding)
)
expire = jwt_attributes.get('exp')

101
config/kube_config_test.py
View File

@ -50,6 +50,10 @@ def _base64(string):
return base64.standard_b64encode(string.encode()).decode()
def _urlsafe_unpadded_b64encode(string):
return base64.urlsafe_b64encode(string.encode()).decode().rstrip('=')
def _format_expiry_datetime(dt):
return dt.strftime(EXPIRY_DATETIME_FORMAT)
@ -97,12 +101,33 @@ TEST_CLIENT_CERT_BASE64 = _base64(TEST_CLIENT_CERT)
TEST_OIDC_TOKEN = "test-oidc-token"
TEST_OIDC_INFO = "{\"name\": \"test\"}"
TEST_OIDC_BASE = _base64(TEST_OIDC_TOKEN) + "." + _base64(TEST_OIDC_INFO)
TEST_OIDC_LOGIN = TEST_OIDC_BASE + "." + TEST_CLIENT_CERT_BASE64
TEST_OIDC_BASE = ".".join([
_urlsafe_unpadded_b64encode(TEST_OIDC_TOKEN),
_urlsafe_unpadded_b64encode(TEST_OIDC_INFO)
])
TEST_OIDC_LOGIN = ".".join([
TEST_OIDC_BASE,
_urlsafe_unpadded_b64encode(TEST_CLIENT_CERT_BASE64)
])
TEST_OIDC_TOKEN = "Bearer %s" % TEST_OIDC_LOGIN
TEST_OIDC_EXP = "{\"name\": \"test\",\"exp\": 536457600}"
TEST_OIDC_EXP_BASE = _base64(TEST_OIDC_TOKEN) + "." + _base64(TEST_OIDC_EXP)
TEST_OIDC_EXPIRED_LOGIN = TEST_OIDC_EXP_BASE + "." + TEST_CLIENT_CERT_BASE64
TEST_OIDC_EXP_BASE = _urlsafe_unpadded_b64encode(
TEST_OIDC_TOKEN) + "." + _urlsafe_unpadded_b64encode(TEST_OIDC_EXP)
TEST_OIDC_EXPIRED_LOGIN = ".".join([
TEST_OIDC_EXP_BASE,
_urlsafe_unpadded_b64encode(TEST_CLIENT_CERT)
])
TEST_OIDC_CONTAINS_RESERVED_CHARACTERS = ".".join([
_urlsafe_unpadded_b64encode(TEST_OIDC_TOKEN),
_urlsafe_unpadded_b64encode(TEST_OIDC_INFO).replace("a", "+"),
_urlsafe_unpadded_b64encode(TEST_CLIENT_CERT)
])
TEST_OIDC_INVALID_PADDING_LENGTH = ".".join([
_urlsafe_unpadded_b64encode(TEST_OIDC_TOKEN),
"aaaaa",
_urlsafe_unpadded_b64encode(TEST_CLIENT_CERT)
])
TEST_OIDC_CA = _base64(TEST_CERTIFICATE_AUTH)
@ -409,6 +434,22 @@ class TestKubeConfigLoader(BaseTestCase):
"user": "expired_oidc_nocert"
}
},
{
"name": "oidc_contains_reserved_character",
"context": {
"cluster": "default",
"user": "oidc_contains_reserved_character"
}
},
{
"name": "oidc_invalid_padding_length",
"context": {
"cluster": "default",
"user": "oidc_invalid_padding_length"
}
},
{
"name": "user_pass",
"context": {
@ -595,6 +636,38 @@ class TestKubeConfigLoader(BaseTestCase):
}
}
},
{
"name": "oidc_contains_reserved_character",
"user": {
"auth-provider": {
"name": "oidc",
"config": {
"client-id": "tectonic-kubectl",
"client-secret": "FAKE_SECRET",
"id-token": TEST_OIDC_CONTAINS_RESERVED_CHARACTERS,
"idp-issuer-url": "https://example.org/identity",
"refresh-token":
"lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"
}
}
}
},
{
"name": "oidc_invalid_padding_length",
"user": {
"auth-provider": {
"name": "oidc",
"config": {
"client-id": "tectonic-kubectl",
"client-secret": "FAKE_SECRET",
"id-token": TEST_OIDC_INVALID_PADDING_LENGTH,
"idp-issuer-url": "https://example.org/identity",
"refresh-token":
"lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"
}
}
}
},
{
"name": "user_pass",
"user": {
@ -793,6 +866,26 @@ class TestKubeConfigLoader(BaseTestCase):
self.assertTrue(loader._load_auth_provider_token())
self.assertEqual("Bearer abc123", loader.token)
def test_oidc_fails_if_contains_reserved_chars(self):
loader = KubeConfigLoader(
config_dict=self.TEST_KUBE_CONFIG,
active_context="oidc_contains_reserved_character",
)
self.assertEqual(
loader._load_oid_token("oidc_contains_reserved_character"),
None,
)
def test_oidc_fails_if_invalid_padding_length(self):
loader = KubeConfigLoader(
config_dict=self.TEST_KUBE_CONFIG,
active_context="oidc_invalid_padding_length",
)
self.assertEqual(
loader._load_oid_token("oidc_invalid_padding_length"),
None,
)
def test_user_pass(self):
expected = FakeConfig(host=TEST_HOST, token=TEST_BASIC_TOKEN)
actual = FakeConfig()