mirror of
https://github.com/microsoft/onefuzz.git
synced 2025-06-18 04:38:09 +00:00
Fix issue when deserializing GithubIssueTemplate with hidden secret (#1008)
This commit is contained in:
@ -14,6 +14,9 @@ from onefuzztypes.job_templates import (
|
|||||||
JobTemplateNotification,
|
JobTemplateNotification,
|
||||||
)
|
)
|
||||||
from onefuzztypes.models import (
|
from onefuzztypes.models import (
|
||||||
|
ADOTemplate,
|
||||||
|
GithubAuth,
|
||||||
|
GithubIssueTemplate,
|
||||||
JobConfig,
|
JobConfig,
|
||||||
Notification,
|
Notification,
|
||||||
NotificationConfig,
|
NotificationConfig,
|
||||||
@ -27,13 +30,14 @@ from onefuzztypes.requests import NotificationCreate
|
|||||||
from __app__.onefuzzlib.orm import hide_secrets
|
from __app__.onefuzzlib.orm import hide_secrets
|
||||||
|
|
||||||
|
|
||||||
|
def hider(secret_data: SecretData) -> SecretData:
|
||||||
|
if not isinstance(secret_data.secret, SecretAddress):
|
||||||
|
secret_data.secret = SecretAddress(url="blah blah")
|
||||||
|
return secret_data
|
||||||
|
|
||||||
|
|
||||||
class TestSecret(unittest.TestCase):
|
class TestSecret(unittest.TestCase):
|
||||||
def test_hide(self) -> None:
|
def test_hide(self) -> None:
|
||||||
def hider(secret_data: SecretData) -> SecretData:
|
|
||||||
if not isinstance(secret_data.secret, SecretAddress):
|
|
||||||
secret_data.secret = SecretAddress(url="blah blah")
|
|
||||||
return secret_data
|
|
||||||
|
|
||||||
notification = Notification(
|
notification = Notification(
|
||||||
container=Container("data"),
|
container=Container("data"),
|
||||||
config=TeamsTemplate(url=SecretData(secret="http://test")),
|
config=TeamsTemplate(url=SecretData(secret="http://test")),
|
||||||
@ -47,11 +51,6 @@ class TestSecret(unittest.TestCase):
|
|||||||
self.fail(f"Invalid config type {type(notification.config)}")
|
self.fail(f"Invalid config type {type(notification.config)}")
|
||||||
|
|
||||||
def test_hide_nested_list(self) -> None:
|
def test_hide_nested_list(self) -> None:
|
||||||
def hider(secret_data: SecretData) -> SecretData:
|
|
||||||
if not isinstance(secret_data.secret, SecretAddress):
|
|
||||||
secret_data.secret = SecretAddress(url="blah blah")
|
|
||||||
return secret_data
|
|
||||||
|
|
||||||
job_template_index = JobTemplateIndex(
|
job_template_index = JobTemplateIndex(
|
||||||
name="test",
|
name="test",
|
||||||
template=JobTemplate(
|
template=JobTemplate(
|
||||||
@ -100,12 +99,33 @@ class TestSecret(unittest.TestCase):
|
|||||||
f"{current_path}"
|
f"{current_path}"
|
||||||
+ "/../../../contrib/onefuzz-job-github-actions/github-issues.json"
|
+ "/../../../contrib/onefuzz-job-github-actions/github-issues.json"
|
||||||
) as json_file:
|
) as json_file:
|
||||||
b = json.load(json_file)
|
notification_dict = json.load(json_file)
|
||||||
b["container"] = "testing"
|
notification_dict["container"] = "testing"
|
||||||
c = NotificationCreate.parse_obj(b)
|
notification1 = NotificationCreate.parse_obj(notification_dict)
|
||||||
d = c.json()
|
|
||||||
e = json.loads(d)
|
assert isinstance(notification1.config, GithubIssueTemplate)
|
||||||
NotificationCreate.parse_obj(e)
|
self.assertIsInstance(
|
||||||
|
notification1.config.auth.secret, GithubAuth, "Invalid secret type"
|
||||||
|
)
|
||||||
|
|
||||||
|
notification2 = NotificationCreate.parse_obj(
|
||||||
|
json.loads(notification1.json())
|
||||||
|
)
|
||||||
|
|
||||||
|
assert isinstance(notification2.config, GithubIssueTemplate)
|
||||||
|
self.assertIsInstance(
|
||||||
|
notification2.config.auth.secret, GithubAuth, "Invalid secret type"
|
||||||
|
)
|
||||||
|
|
||||||
|
hide_secrets(notification2, hider)
|
||||||
|
|
||||||
|
notification3 = NotificationCreate.parse_obj(
|
||||||
|
json.loads(notification2.json())
|
||||||
|
)
|
||||||
|
assert isinstance(notification3.config, GithubIssueTemplate)
|
||||||
|
self.assertIsInstance(
|
||||||
|
notification3.config.auth.secret, SecretAddress, "Invalid secret type"
|
||||||
|
)
|
||||||
|
|
||||||
def test_roundtrip_team_issue(self) -> None:
|
def test_roundtrip_team_issue(self) -> None:
|
||||||
a = """
|
a = """
|
||||||
@ -115,11 +135,28 @@ class TestSecret(unittest.TestCase):
|
|||||||
}
|
}
|
||||||
|
|
||||||
""" # noqa
|
""" # noqa
|
||||||
b = json.loads(a)
|
notification_dict = json.loads(a)
|
||||||
c = NotificationCreate.parse_obj(b)
|
notification_dict["container"] = "testing"
|
||||||
d = c.json()
|
notification1 = NotificationCreate.parse_obj(notification_dict)
|
||||||
e = json.loads(d)
|
|
||||||
NotificationCreate.parse_obj(e)
|
assert isinstance(notification1.config, TeamsTemplate)
|
||||||
|
self.assertIsInstance(
|
||||||
|
notification1.config.url.secret, str, "Invalid secret type"
|
||||||
|
)
|
||||||
|
|
||||||
|
notification2 = NotificationCreate.parse_obj(json.loads(notification1.json()))
|
||||||
|
assert isinstance(notification2.config, TeamsTemplate)
|
||||||
|
self.assertIsInstance(
|
||||||
|
notification2.config.url.secret, str, "Invalid secret type"
|
||||||
|
)
|
||||||
|
|
||||||
|
hide_secrets(notification2, hider)
|
||||||
|
|
||||||
|
notification3 = NotificationCreate.parse_obj(json.loads(notification2.json()))
|
||||||
|
assert isinstance(notification3.config, TeamsTemplate)
|
||||||
|
self.assertIsInstance(
|
||||||
|
notification3.config.url.secret, SecretAddress, "Invalid secret type"
|
||||||
|
)
|
||||||
|
|
||||||
def test_roundtrip_ado(self) -> None:
|
def test_roundtrip_ado(self) -> None:
|
||||||
current_path = pathlib.Path(__file__).parent.absolute()
|
current_path = pathlib.Path(__file__).parent.absolute()
|
||||||
@ -127,9 +164,30 @@ class TestSecret(unittest.TestCase):
|
|||||||
f"{current_path}"
|
f"{current_path}"
|
||||||
+ "/../../../contrib/onefuzz-job-azure-devops-pipeline/ado-work-items.json" # noqa
|
+ "/../../../contrib/onefuzz-job-azure-devops-pipeline/ado-work-items.json" # noqa
|
||||||
) as json_file:
|
) as json_file:
|
||||||
b = json.load(json_file)
|
notification_dict = json.load(json_file)
|
||||||
b["container"] = "testing"
|
notification_dict["container"] = "testing"
|
||||||
c = NotificationCreate.parse_obj(b)
|
notification1 = NotificationCreate.parse_obj(notification_dict)
|
||||||
d = c.json()
|
assert isinstance(notification1.config, ADOTemplate)
|
||||||
e = json.loads(d)
|
self.assertIsInstance(
|
||||||
NotificationCreate.parse_obj(e)
|
notification1.config.auth_token.secret, str, "Invalid secret type"
|
||||||
|
)
|
||||||
|
|
||||||
|
notification2 = NotificationCreate.parse_obj(
|
||||||
|
json.loads(notification1.json())
|
||||||
|
)
|
||||||
|
assert isinstance(notification2.config, ADOTemplate)
|
||||||
|
self.assertIsInstance(
|
||||||
|
notification2.config.auth_token.secret, str, "Invalid secret type"
|
||||||
|
)
|
||||||
|
|
||||||
|
hide_secrets(notification2, hider)
|
||||||
|
|
||||||
|
notification3 = NotificationCreate.parse_obj(
|
||||||
|
json.loads(notification2.json())
|
||||||
|
)
|
||||||
|
assert isinstance(notification3.config, ADOTemplate)
|
||||||
|
self.assertIsInstance(
|
||||||
|
notification3.config.auth_token.secret,
|
||||||
|
SecretAddress,
|
||||||
|
"Invalid secret type",
|
||||||
|
)
|
||||||
|
@ -511,15 +511,26 @@ class GithubIssueTemplate(BaseModel):
|
|||||||
# validator needed for backward compatibility
|
# validator needed for backward compatibility
|
||||||
@validator("auth", pre=True, always=True)
|
@validator("auth", pre=True, always=True)
|
||||||
def validate_auth(cls, v: Any) -> SecretData:
|
def validate_auth(cls, v: Any) -> SecretData:
|
||||||
|
def try_parse_GithubAuth(x: dict) -> Optional[GithubAuth]:
|
||||||
|
try:
|
||||||
|
return GithubAuth.parse_obj(x)
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
|
||||||
if isinstance(v, GithubAuth):
|
if isinstance(v, GithubAuth):
|
||||||
return SecretData(secret=v)
|
return SecretData(secret=v)
|
||||||
elif isinstance(v, SecretData):
|
elif isinstance(v, SecretData):
|
||||||
return v
|
return v
|
||||||
elif isinstance(v, dict):
|
elif isinstance(v, dict):
|
||||||
try:
|
githubAuth = try_parse_GithubAuth(v)
|
||||||
return SecretData(GithubAuth.parse_obj(v))
|
if githubAuth:
|
||||||
except Exception:
|
return SecretData(secret=githubAuth)
|
||||||
return SecretData(GithubAuth.parse_obj(v["secret"]))
|
|
||||||
|
githubAuth = try_parse_GithubAuth(v["secret"])
|
||||||
|
if githubAuth:
|
||||||
|
return SecretData(secret=githubAuth)
|
||||||
|
|
||||||
|
return SecretData(secret=v["secret"])
|
||||||
else:
|
else:
|
||||||
raise TypeError(f"invalid datatype {type(v)}")
|
raise TypeError(f"invalid datatype {type(v)}")
|
||||||
|
|
||||||
|
@ -6,8 +6,8 @@
|
|||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
from onefuzztypes.models import Scaleset, SecretData, TeamsTemplate
|
from onefuzztypes.models import Scaleset, SecretData, TeamsTemplate
|
||||||
from onefuzztypes.requests import NotificationCreate
|
|
||||||
from onefuzztypes.primitives import PoolName, Region
|
from onefuzztypes.primitives import PoolName, Region
|
||||||
|
from onefuzztypes.requests import NotificationCreate
|
||||||
from pydantic import ValidationError
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user