Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add ecr image scan (#4166)

Browse Source 
* Add ecr.start_image_scan

* Add ecr.describe_image_scan_findings
This commit is contained in:
Anton Grübel 2021-08-12 14:06:21 +09:00 committed by GitHub
parent a5eb46962d
commit b8405b39b5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 435 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
moto/ecr/exceptions.py
View File

@ -16,6 +16,16 @@ class LifecyclePolicyNotFoundException(JsonRESTError):
)
class LimitExceededException(JsonRESTError):
code = 400
def __init__(self):
super().__init__(
error_type=__class__.__name__,
message=("The scan quota per image has been exceeded. Wait and try again."),
)
class RegistryPolicyNotFoundException(JsonRESTError):
code = 400
@ -101,3 +111,17 @@ class InvalidParameterException(JsonRESTError):
def __init__(self, message):
super().__init__(error_type=__class__.__name__, message=message)
class ScanNotFoundException(JsonRESTError):
code = 400
def __init__(self, image_id, repository_name, registry_id):
super().__init__(
error_type=__class__.__name__,
message=(
f"Image scan does not exist for the image with '{image_id}' "
f"in the repository with name '{repository_name}' "
f"in the registry with id '{registry_id}'"
),
)

144
moto/ecr/models.py
View File

@ -7,7 +7,7 @@ import uuid
from collections import namedtuple
from datetime import datetime
from random import random
from typing import Dict
from typing import Dict, List
from botocore.exceptions import ParamValidationError
@ -23,6 +23,8 @@ from moto.ecr.exceptions import (
RepositoryPolicyNotFoundException,
LifecyclePolicyNotFoundException,
RegistryPolicyNotFoundException,
LimitExceededException,
ScanNotFoundException,
)
from moto.ecr.policy_validation import EcrLifecyclePolicyValidator
from moto.iam.exceptions import MalformedPolicyDocument
@ -87,7 +89,7 @@ class Repository(BaseObject, CloudFormationModel):
)
self.policy = None
self.lifecycle_policy = None
self.images = []
self.images: List[Image] = []
def _determine_encryption_config(self, encryption_config):
if not encryption_config:
@ -98,6 +100,31 @@ class Repository(BaseObject, CloudFormationModel):
] = f"arn:aws:kms:{self.region_name}:{ACCOUNT_ID}:key/{uuid.uuid4()}"
return encryption_config
def _get_image(self, image_tag, image_digest):
# you can either search for one or both
image = next(
(
i
for i in self.images
if (not image_tag or image_tag in i.image_tags)
and (not image_digest or image_digest == i.get_image_digest())
),
None,
)
if not image:
image_id_rep = "{{imageDigest:'{0}', imageTag:'{1}'}}".format(
image_digest or "null", image_tag or "null"
)
raise ImageNotFoundException(
image_id=image_id_rep,
repository_name=self.name,
registry_id=self.registry_id,
)
return image
@property
def physical_resource_id(self):
return self.name
@ -210,6 +237,7 @@ class Image(BaseObject):
self.registry_id = registry_id
self.image_digest = digest
self.image_pushed_at = str(datetime.utcnow().isoformat())
self.last_scan = None
def _create_digest(self):
image_contents = "docker_image{0}".format(int(random() * 10 ** 6))
@ -407,38 +435,15 @@ class ECRBackend(BaseBackend):
return images
def describe_images(self, repository_name, registry_id=None, image_ids=None):
if repository_name in self.repositories:
repository = self.repositories[repository_name]
else:
raise RepositoryNotFoundException(
repository_name, registry_id or DEFAULT_REGISTRY_ID
)
repository = self._get_repository(repository_name, registry_id)
if image_ids:
response = set()
for image_id in image_ids:
found = False
for image in repository.images:
if (
"imageDigest" in image_id
and image.get_image_digest() == image_id["imageDigest"]
) or (
"imageTag" in image_id
and image_id["imageTag"] in image.image_tags
):
found = True
response.add(image)
if not found:
image_id_representation = "{imageDigest:'%s', imageTag:'%s'}" % (
image_id.get("imageDigest", "null"),
image_id.get("imageTag", "null"),
)
raise ImageNotFoundException(
image_id=image_id_representation,
repository_name=repository_name,
registry_id=registry_id or DEFAULT_REGISTRY_ID,
)
response = set(
repository._get_image(
image_id.get("imageTag"), image_id.get("imageDigest")
)
for image_id in image_ids
)
else:
response = []
@ -816,6 +821,81 @@ class ECRBackend(BaseBackend):
"policyText": policy,
}
def start_image_scan(self, registry_id, repository_name, image_id):
repo = self._get_repository(repository_name, registry_id)
image = repo._get_image(image_id.get("imageTag"), image_id.get("imageDigest"))
# scanning an image is only allowed once per day
if image.last_scan and image.last_scan.date() == datetime.today().date():
raise LimitExceededException()
image.last_scan = datetime.today()
return {
"registryId": repo.registry_id,
"repositoryName": repository_name,
"imageId": {
"imageDigest": image.image_digest,
"imageTag": image.image_tag,
},
"imageScanStatus": {"status": "IN_PROGRESS"},
}
def describe_image_scan_findings(self, registry_id, repository_name, image_id):
repo = self._get_repository(repository_name, registry_id)
image = repo._get_image(image_id.get("imageTag"), image_id.get("imageDigest"))
if not image.last_scan:
image_id_rep = "{{imageDigest:'{0}', imageTag:'{1}'}}".format(
image_id.get("imageDigest") or "null",
image_id.get("imageTag") or "null",
)
raise ScanNotFoundException(
image_id=image_id_rep,
repository_name=repository_name,
registry_id=repo.registry_id,
)
return {
"registryId": repo.registry_id,
"repositoryName": repository_name,
"imageId": {
"imageDigest": image.image_digest,
"imageTag": image.image_tag,
},
"imageScanStatus": {
"status": "COMPLETE",
"description": "The scan was completed successfully.",
},
"imageScanFindings": {
"imageScanCompletedAt": iso_8601_datetime_without_milliseconds(
image.last_scan
),
"vulnerabilitySourceUpdatedAt": iso_8601_datetime_without_milliseconds(
datetime.utcnow()
),
"findings": [
{
"name": "CVE-9999-9999",
"uri": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-9999-9999",
"severity": "HIGH",
"attributes": [
{"key": "package_version", "value": "9.9.9"},
{"key": "package_name", "value": "moto_fake"},
{
"key": "CVSS2_VECTOR",
"value": "AV:N/AC:L/Au:N/C:P/I:P/A:P",
},
{"key": "CVSS2_SCORE", "value": "7.5"},
],
}
],
"findingSeverityCounts": {"HIGH": 1},
},
}
ecr_backends = {}
for region, ec2_backend in ec2_backends.items():

26
moto/ecr/responses.py
View File

@ -271,3 +271,29 @@ class ECRResponse(BaseResponse):
def delete_registry_policy(self):
return json.dumps(self.ecr_backend.delete_registry_policy())
def start_image_scan(self):
registry_id = self._get_param("registryId")
repository_name = self._get_param("repositoryName")
image_id = self._get_param("imageId")
return json.dumps(
self.ecr_backend.start_image_scan(
registry_id=registry_id,
repository_name=repository_name,
image_id=image_id,
)
)
def describe_image_scan_findings(self):
registry_id = self._get_param("registryId")
repository_name = self._get_param("repositoryName")
image_id = self._get_param("imageId")
return json.dumps(
self.ecr_backend.describe_image_scan_findings(
registry_id=registry_id,
repository_name=repository_name,
image_id=image_id,
)
)

277
tests/test_ecr/test_ecr_boto3.py
View File

@ -651,24 +651,24 @@ def test_describe_image_that_doesnt_exist():
error_msg1 = re.compile(
r".*The image with imageId {imageDigest:'null', imageTag:'testtag'} does not exist within "
r"the repository with name 'test_repository' in the registry with id '123'.*",
r"the repository with name 'test_repository' in the registry with id '123456789012'.*",
re.MULTILINE,
)
client.describe_images.when.called_with(
repositoryName="test_repository",
imageIds=[{"imageTag": "testtag"}],
registryId="123",
registryId=ACCOUNT_ID,
).should.throw(client.exceptions.ImageNotFoundException, error_msg1)
error_msg2 = re.compile(
r".*The repository with name 'repo-that-doesnt-exist' does not exist in the registry with id '123'.*",
r".*The repository with name 'repo-that-doesnt-exist' does not exist in the registry with id '123456789012'.*",
re.MULTILINE,
)
client.describe_images.when.called_with(
repositoryName="repo-that-doesnt-exist",
imageIds=[{"imageTag": "testtag"}],
registryId="123",
registryId=ACCOUNT_ID,
).should.throw(ClientError, error_msg2)
@ -2113,3 +2113,272 @@ def test_delete_registry_policy_error_policy_not_exists():
ex.response["Error"]["Message"].should.equal(
f"Registry policy does not exist in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_start_image_scan():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
image_tag = "latest"
image_digest = client.put_image(
repositoryName=repo_name,
imageManifest=json.dumps(_create_image_manifest()),
imageTag="latest",
)["image"]["imageId"]["imageDigest"]
# when
response = client.start_image_scan(
repositoryName=repo_name, imageId={"imageTag": image_tag}
)
# then
response["registryId"].should.equal(ACCOUNT_ID)
response["repositoryName"].should.equal(repo_name)
response["imageId"].should.equal(
{"imageDigest": image_digest, "imageTag": image_tag}
)
response["imageScanStatus"].should.equal({"status": "IN_PROGRESS"})
@mock_ecr
def test_start_image_scan_error_repo_not_exists():
# given
region_name = "eu-central-1"
client = boto3.client("ecr", region_name=region_name)
repo_name = "not-exists"
# when
with pytest.raises(ClientError) as e:
client.start_image_scan(
repositoryName=repo_name, imageId={"imageTag": "latest"}
)
# then
ex = e.value
ex.operation_name.should.equal("StartImageScan")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"The repository with name '{repo_name}' does not exist "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_start_image_scan_error_image_not_exists():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
image_tag = "not-exists"
# when
with pytest.raises(ClientError) as e:
client.start_image_scan(
repositoryName=repo_name, imageId={"imageTag": image_tag}
)
# then
ex = e.value
ex.operation_name.should.equal("StartImageScan")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ImageNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"The image with imageId {{imageDigest:'null', imageTag:'{image_tag}'}} does not exist "
f"within the repository with name '{repo_name}' "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_start_image_scan_error_image_tag_digest_mismatch():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
image_digest = client.put_image(
repositoryName=repo_name,
imageManifest=json.dumps(_create_image_manifest()),
imageTag="latest",
)["image"]["imageId"]["imageDigest"]
image_tag = "not-latest"
# when
with pytest.raises(ClientError) as e:
client.start_image_scan(
repositoryName=repo_name,
imageId={"imageTag": image_tag, "imageDigest": image_digest},
)
# then
ex = e.value
ex.operation_name.should.equal("StartImageScan")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ImageNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"The image with imageId {{imageDigest:'{image_digest}', imageTag:'{image_tag}'}} does not exist "
f"within the repository with name '{repo_name}' "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_start_image_scan_error_daily_limit():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
image_tag = "latest"
image_digest = client.put_image(
repositoryName=repo_name,
imageManifest=json.dumps(_create_image_manifest()),
imageTag="latest",
)["image"]["imageId"]["imageDigest"]
client.start_image_scan(repositoryName=repo_name, imageId={"imageTag": image_tag})
# when
with pytest.raises(ClientError) as e:
client.start_image_scan(
repositoryName=repo_name, imageId={"imageTag": image_tag}
)
# then
ex = e.value
ex.operation_name.should.equal("StartImageScan")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("LimitExceededException")
ex.response["Error"]["Message"].should.equal(
"The scan quota per image has been exceeded. Wait and try again."
)
@mock_ecr
def test_describe_image_scan_findings():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
image_tag = "latest"
image_digest = client.put_image(
repositoryName=repo_name,
imageManifest=json.dumps(_create_image_manifest()),
imageTag="latest",
)["image"]["imageId"]["imageDigest"]
client.start_image_scan(repositoryName=repo_name, imageId={"imageTag": image_tag})
# when
response = client.describe_image_scan_findings(
repositoryName=repo_name, imageId={"imageTag": image_tag}
)
# then
response["registryId"].should.equal(ACCOUNT_ID)
response["repositoryName"].should.equal(repo_name)
response["imageId"].should.equal(
{"imageDigest": image_digest, "imageTag": image_tag}
)
response["imageScanStatus"].should.equal(
{"status": "COMPLETE", "description": "The scan was completed successfully."}
)
scan_findings = response["imageScanFindings"]
scan_findings["imageScanCompletedAt"].should.be.a(datetime)
scan_findings["vulnerabilitySourceUpdatedAt"].should.be.a(datetime)
scan_findings["findings"].should.equal(
[
{
"name": "CVE-9999-9999",
"uri": "https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-9999-9999",
"severity": "HIGH",
"attributes": [
{"key": "package_version", "value": "9.9.9"},
{"key": "package_name", "value": "moto_fake"},
{"key": "CVSS2_VECTOR", "value": "AV:N/AC:L/Au:N/C:P/I:P/A:P",},
{"key": "CVSS2_SCORE", "value": "7.5"},
],
}
]
)
scan_findings["findingSeverityCounts"].should.equal({"HIGH": 1})
@mock_ecr
def test_describe_image_scan_findings_error_repo_not_exists():
# given
region_name = "eu-central-1"
client = boto3.client("ecr", region_name=region_name)
repo_name = "not-exists"
# when
with pytest.raises(ClientError) as e:
client.describe_image_scan_findings(
repositoryName=repo_name, imageId={"imageTag": "latest"}
)
# then
ex = e.value
ex.operation_name.should.equal("DescribeImageScanFindings")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("RepositoryNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"The repository with name '{repo_name}' does not exist "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_describe_image_scan_findings_error_image_not_exists():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
image_tag = "not-exists"
# when
with pytest.raises(ClientError) as e:
client.describe_image_scan_findings(
repositoryName=repo_name, imageId={"imageTag": image_tag}
)
# then
ex = e.value
ex.operation_name.should.equal("DescribeImageScanFindings")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ImageNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"The image with imageId {{imageDigest:'null', imageTag:'{image_tag}'}} does not exist "
f"within the repository with name '{repo_name}' "
f"in the registry with id '{ACCOUNT_ID}'"
)
@mock_ecr
def test_describe_image_scan_findings_error_scan_not_exists():
# given
client = boto3.client("ecr", region_name="eu-central-1")
repo_name = "test-repo"
client.create_repository(repositoryName=repo_name)
image_tag = "latest"
client.put_image(
repositoryName=repo_name,
imageManifest=json.dumps(_create_image_manifest()),
imageTag=image_tag,
)
# when
with pytest.raises(ClientError) as e:
client.describe_image_scan_findings(
repositoryName=repo_name, imageId={"imageTag": image_tag}
)
# then
ex = e.value
ex.operation_name.should.equal("DescribeImageScanFindings")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ScanNotFoundException")
ex.response["Error"]["Message"].should.equal(
f"Image scan does not exist for the image with '{{imageDigest:'null', imageTag:'{image_tag}'}}' "
f"in the repository with name '{repo_name}' "
f"in the registry with id '{ACCOUNT_ID}'"
)