From 9f662e6eef4fcba211518121b9ecb180080b7002 Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Sat, 18 Jul 2026 23:29:41 +0900 Subject: [PATCH 1/4] feat(ec2,iam): add operation depth for core services MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add 24 commonly-used operations that were previously unimplemented, all reusing existing SQLite-backed stores. EC2 (17): - Instance lifecycle: StartInstances, StopInstances, RebootInstances - Discovery: DescribeAvailabilityZones, DescribeRegions - Security group rules: AuthorizeSecurityGroupIngress/Egress, RevokeSecurityGroupIngress/Egress (new security_group_rules table), surfaced in DescribeSecurityGroups - Elastic IP: AssociateAddress, DisassociateAddress, ReleaseAddress, DescribeAddresses (adds an association_id column) - Cleanup: DeleteSecurityGroup, DeleteSubnet, DescribeTags, DeleteTags IAM (7) — group policy management mirroring the existing user/role code: - AttachGroupPolicy, DetachGroupPolicy, ListAttachedGroupPolicies, PutGroupPolicy, GetGroupPolicy, DeleteGroupPolicy, ListGroupPolicies (new group_policies + group_inline_policies tables) Covered by boto3 compatibility tests (suite: 729 -> 740). --- internal/services/ec2/provider.go | 458 +++++++++++++++++++++++++++- internal/services/ec2/store.go | 274 ++++++++++++++++- internal/services/iam/provider.go | 171 +++++++++++ internal/services/iam/store.go | 119 ++++++++ tests/compatibility/test_ec2.py | 139 +++++++++ tests/compatibility/test_iam_sts.py | 35 +++ 6 files changed, 1186 insertions(+), 10 deletions(-) diff --git a/internal/services/ec2/provider.go b/internal/services/ec2/provider.go index 002eda74..b5f467e0 100644 --- a/internal/services/ec2/provider.go +++ b/internal/services/ec2/provider.go @@ -69,6 +69,12 @@ func (p *Provider) HandleRequest(ctx context.Context, op string, req *http.Reque return p.handleDescribeInstances(form) case "TerminateInstances": return p.handleTerminateInstances(form) + case "StartInstances": + return p.changeInstanceState(form, "StartInstancesResponse", "running") + case "StopInstances": + return p.changeInstanceState(form, "StopInstancesResponse", "stopped") + case "RebootInstances": + return p.handleRebootInstances(form) case "CreateVpc": return p.handleCreateVpc(form) case "DescribeVpcs": @@ -79,14 +85,42 @@ func (p *Provider) HandleRequest(ctx context.Context, op string, req *http.Reque return p.handleCreateSubnet(form) case "DescribeSubnets": return p.handleDescribeSubnets(form) + case "DeleteSubnet": + return p.handleDeleteSubnet(form) case "CreateSecurityGroup": return p.handleCreateSecurityGroup(form) case "DescribeSecurityGroups": return p.handleDescribeSecurityGroups(form) + case "DeleteSecurityGroup": + return p.handleDeleteSecurityGroup(form) + case "AuthorizeSecurityGroupIngress": + return p.handleAuthorizeSecurityGroup(form, false) + case "AuthorizeSecurityGroupEgress": + return p.handleAuthorizeSecurityGroup(form, true) + case "RevokeSecurityGroupIngress": + return p.handleRevokeSecurityGroup(form, false) + case "RevokeSecurityGroupEgress": + return p.handleRevokeSecurityGroup(form, true) case "CreateTags": return p.handleCreateTags(form) + case "DescribeTags": + return p.handleDescribeTags(form) + case "DeleteTags": + return p.handleDeleteTags(form) case "AllocateAddress": return p.handleAllocateAddress(form) + case "AssociateAddress": + return p.handleAssociateAddress(form) + case "DisassociateAddress": + return p.handleDisassociateAddress(form) + case "ReleaseAddress": + return p.handleReleaseAddress(form) + case "DescribeAddresses": + return p.handleDescribeAddresses(form) + case "DescribeAvailabilityZones": + return p.handleDescribeAvailabilityZones(form) + case "DescribeRegions": + return p.handleDescribeRegions(form) case "CreateVolume": return p.handleCreateVolume(form) case "DescribeVolumes": @@ -365,6 +399,76 @@ func (p *Provider) handleTerminateInstances(form url.Values) (*plugin.Response, return ec2XMLResponse(http.StatusOK, terminateInstancesResponse{InstancesSet: items}) } +// instanceIDsFromForm collects InstanceId.1, InstanceId.2, … from a Query form. +func instanceIDsFromForm(form url.Values) []string { + var ids []string + for i := 1; ; i++ { + id := form.Get(fmt.Sprintf("InstanceId.%d", i)) + if id == "" { + break + } + ids = append(ids, id) + } + return ids +} + +// changeInstanceState handles StartInstances/StopInstances: it sets the new +// state and reports the per-instance previous → current transition. +func (p *Provider) changeInstanceState(form url.Values, respName, newState string) (*plugin.Response, error) { + ids := instanceIDsFromForm(form) + if len(ids) == 0 { + return ec2XMLError("MissingParameter", "at least one InstanceId is required", http.StatusBadRequest), nil + } + prev, err := p.store.SetInstanceState(defaultAccountID, ids, newState) + if err != nil { + if errors.Is(err, ErrInstanceNotFound) { + return ec2XMLError("InvalidInstanceID.NotFound", err.Error(), http.StatusBadRequest), nil + } + return nil, err + } + type stateChangeXML struct { + InstanceId string `xml:"instanceId"` + CurrentState struct { + Name string `xml:"name"` + } `xml:"currentState"` + PreviousState struct { + Name string `xml:"name"` + } `xml:"previousState"` + } + type changeStateResponse struct { + XMLName xml.Name + InstancesSet []stateChangeXML `xml:"instancesSet>item"` + } + items := make([]stateChangeXML, 0, len(ids)) + for _, id := range ids { + sc := stateChangeXML{InstanceId: id} + sc.CurrentState.Name = newState + sc.PreviousState.Name = prev[id] + items = append(items, sc) + } + return ec2XMLResponse(http.StatusOK, changeStateResponse{XMLName: xml.Name{Local: respName}, InstancesSet: items}) +} + +// handleRebootInstances validates the instances exist; reboot does not change +// the reported state (mirrors the real EC2 behaviour). +func (p *Provider) handleRebootInstances(form url.Values) (*plugin.Response, error) { + ids := instanceIDsFromForm(form) + if len(ids) == 0 { + return ec2XMLError("MissingParameter", "at least one InstanceId is required", http.StatusBadRequest), nil + } + if err := p.store.InstancesExist(defaultAccountID, ids); err != nil { + if errors.Is(err, ErrInstanceNotFound) { + return ec2XMLError("InvalidInstanceID.NotFound", err.Error(), http.StatusBadRequest), nil + } + return nil, err + } + type rebootInstancesResponse struct { + XMLName xml.Name `xml:"RebootInstancesResponse"` + Return bool `xml:"return"` + } + return ec2XMLResponse(http.StatusOK, rebootInstancesResponse{Return: true}) +} + func (p *Provider) handleCreateVpc(form url.Values) (*plugin.Response, error) { cidr := form.Get("CidrBlock") if cidr == "" { @@ -495,19 +599,50 @@ func (p *Provider) handleDescribeSecurityGroups(_ url.Values) (*plugin.Response, if err != nil { return nil, err } + type ipRangeXML struct { + CidrIp string `xml:"cidrIp"` + } + type ipPermissionXML struct { + IpProtocol string `xml:"ipProtocol"` + FromPort *int `xml:"fromPort,omitempty"` + ToPort *int `xml:"toPort,omitempty"` + IpRanges []ipRangeXML `xml:"ipRanges>item,omitempty"` + } type sgItemXML struct { - GroupId string `xml:"groupId"` - GroupName string `xml:"groupName"` - VpcId string `xml:"vpcId"` - Description string `xml:"groupDescription"` + GroupId string `xml:"groupId"` + GroupName string `xml:"groupName"` + VpcId string `xml:"vpcId"` + Description string `xml:"groupDescription"` + IpPermissions []ipPermissionXML `xml:"ipPermissions>item,omitempty"` + IpPermissionsEgress []ipPermissionXML `xml:"ipPermissionsEgress>item,omitempty"` } type describeSecurityGroupsResponse struct { XMLName xml.Name `xml:"DescribeSecurityGroupsResponse"` SecurityGroupInfo []sgItemXML `xml:"securityGroupInfo>item"` } + toPerm := func(r SecurityGroupRule) ipPermissionXML { + from, to := r.FromPort, r.ToPort + perm := ipPermissionXML{IpProtocol: r.Protocol, FromPort: &from, ToPort: &to} + if r.CIDR != "" { + perm.IpRanges = []ipRangeXML{{CidrIp: r.CIDR}} + } + return perm + } items := make([]sgItemXML, 0, len(sgs)) for _, sg := range sgs { - items = append(items, sgItemXML{GroupId: sg.GroupID, GroupName: sg.GroupName, VpcId: sg.VpcID, Description: sg.Description}) + item := sgItemXML{GroupId: sg.GroupID, GroupName: sg.GroupName, VpcId: sg.VpcID, Description: sg.Description} + rules, err := p.store.SecurityGroupRules(defaultAccountID, sg.GroupID) + if err != nil { + return nil, err + } + for _, r := range rules { + if r.Egress { + item.IpPermissionsEgress = append(item.IpPermissionsEgress, toPerm(r)) + } else { + item.IpPermissions = append(item.IpPermissions, toPerm(r)) + } + } + items = append(items, item) } return ec2XMLResponse(http.StatusOK, describeSecurityGroupsResponse{SecurityGroupInfo: items}) } @@ -562,6 +697,319 @@ func (p *Provider) handleAllocateAddress(form url.Values) (*plugin.Response, err }) } +func (p *Provider) handleAssociateAddress(form url.Values) (*plugin.Response, error) { + allocationID := form.Get("AllocationId") + instanceID := form.Get("InstanceId") + if allocationID == "" || instanceID == "" { + return ec2XMLError("MissingParameter", "AllocationId and InstanceId are required", http.StatusBadRequest), nil + } + assocID, err := p.store.AssociateAddress(defaultAccountID, allocationID, instanceID) + if err != nil { + if errors.Is(err, ErrAddressNotFound) { + return ec2XMLError("InvalidAllocationID.NotFound", err.Error(), http.StatusBadRequest), nil + } + return nil, err + } + type associateAddressResponse struct { + XMLName xml.Name `xml:"AssociateAddressResponse"` + Return bool `xml:"return"` + AssociationId string `xml:"associationId"` + } + return ec2XMLResponse(http.StatusOK, associateAddressResponse{Return: true, AssociationId: assocID}) +} + +func (p *Provider) handleDisassociateAddress(form url.Values) (*plugin.Response, error) { + assocID := form.Get("AssociationId") + if assocID == "" { + return ec2XMLError("MissingParameter", "AssociationId is required", http.StatusBadRequest), nil + } + if err := p.store.DisassociateAddress(defaultAccountID, assocID); err != nil { + if errors.Is(err, ErrAddressNotFound) { + return ec2XMLError("InvalidAssociationID.NotFound", err.Error(), http.StatusBadRequest), nil + } + return nil, err + } + type disassociateAddressResponse struct { + XMLName xml.Name `xml:"DisassociateAddressResponse"` + Return bool `xml:"return"` + } + return ec2XMLResponse(http.StatusOK, disassociateAddressResponse{Return: true}) +} + +func (p *Provider) handleReleaseAddress(form url.Values) (*plugin.Response, error) { + allocationID := form.Get("AllocationId") + if allocationID == "" { + return ec2XMLError("MissingParameter", "AllocationId is required", http.StatusBadRequest), nil + } + if err := p.store.ReleaseAddress(defaultAccountID, allocationID); err != nil { + if errors.Is(err, ErrAddressNotFound) { + return ec2XMLError("InvalidAllocationID.NotFound", err.Error(), http.StatusBadRequest), nil + } + return nil, err + } + type releaseAddressResponse struct { + XMLName xml.Name `xml:"ReleaseAddressResponse"` + Return bool `xml:"return"` + } + return ec2XMLResponse(http.StatusOK, releaseAddressResponse{Return: true}) +} + +func (p *Provider) handleDescribeAddresses(_ url.Values) (*plugin.Response, error) { + addrs, err := p.store.DescribeAddresses(defaultAccountID) + if err != nil { + return nil, err + } + type addrXML struct { + PublicIp string `xml:"publicIp"` + AllocationId string `xml:"allocationId"` + Domain string `xml:"domain"` + InstanceId string `xml:"instanceId,omitempty"` + AssociationId string `xml:"associationId,omitempty"` + } + type describeAddressesResponse struct { + XMLName xml.Name `xml:"DescribeAddressesResponse"` + AddressesSet []addrXML `xml:"addressesSet>item"` + } + items := make([]addrXML, 0, len(addrs)) + for _, a := range addrs { + items = append(items, addrXML{ + PublicIp: a.PublicIP, AllocationId: a.AllocationID, Domain: a.Domain, + InstanceId: a.InstanceID, AssociationId: a.AssociationID, + }) + } + return ec2XMLResponse(http.StatusOK, describeAddressesResponse{AddressesSet: items}) +} + +func (p *Provider) handleDescribeAvailabilityZones(_ url.Values) (*plugin.Response, error) { + type azXML struct { + ZoneName string `xml:"zoneName"` + ZoneId string `xml:"zoneId"` + ZoneState string `xml:"zoneState"` + RegionName string `xml:"regionName"` + } + type describeAvailabilityZonesResponse struct { + XMLName xml.Name `xml:"DescribeAvailabilityZonesResponse"` + AvailabilityZoneInfo []azXML `xml:"availabilityZoneInfo>item"` + } + const region = "us-east-1" + suffixes := []string{"a", "b", "c", "d", "f"} + items := make([]azXML, 0, len(suffixes)) + for i, sfx := range suffixes { + items = append(items, azXML{ + ZoneName: region + sfx, ZoneId: fmt.Sprintf("use1-az%d", i+1), + ZoneState: "available", RegionName: region, + }) + } + return ec2XMLResponse(http.StatusOK, describeAvailabilityZonesResponse{AvailabilityZoneInfo: items}) +} + +func (p *Provider) handleDescribeRegions(_ url.Values) (*plugin.Response, error) { + type regionXML struct { + RegionName string `xml:"regionName"` + RegionEndpoint string `xml:"regionEndpoint"` + } + type describeRegionsResponse struct { + XMLName xml.Name `xml:"DescribeRegionsResponse"` + RegionInfo []regionXML `xml:"regionInfo>item"` + } + regions := []string{ + "us-east-1", "us-east-2", "us-west-1", "us-west-2", + "eu-west-1", "eu-central-1", "ap-southeast-1", "ap-southeast-2", "ap-northeast-1", + } + items := make([]regionXML, 0, len(regions)) + for _, r := range regions { + items = append(items, regionXML{RegionName: r, RegionEndpoint: "ec2." + r + ".amazonaws.com"}) + } + return ec2XMLResponse(http.StatusOK, describeRegionsResponse{RegionInfo: items}) +} + +func (p *Provider) handleDeleteSubnet(form url.Values) (*plugin.Response, error) { + id := form.Get("SubnetId") + if id == "" { + return ec2XMLError("MissingParameter", "SubnetId is required", http.StatusBadRequest), nil + } + if err := p.store.DeleteSubnet(defaultAccountID, id); err != nil { + if errors.Is(err, ErrSubnetNotFound) { + return ec2XMLError("InvalidSubnetID.NotFound", err.Error(), http.StatusBadRequest), nil + } + return nil, err + } + type deleteSubnetResponse struct { + XMLName xml.Name `xml:"DeleteSubnetResponse"` + Return bool `xml:"return"` + } + return ec2XMLResponse(http.StatusOK, deleteSubnetResponse{Return: true}) +} + +func (p *Provider) handleDeleteSecurityGroup(form url.Values) (*plugin.Response, error) { + id := form.Get("GroupId") + if id == "" { + id = form.Get("GroupName") + } + if id == "" { + return ec2XMLError("MissingParameter", "GroupId or GroupName is required", http.StatusBadRequest), nil + } + if err := p.store.DeleteSecurityGroup(defaultAccountID, id); err != nil { + if errors.Is(err, ErrSecurityGroupNotFound) { + return ec2XMLError("InvalidGroup.NotFound", err.Error(), http.StatusBadRequest), nil + } + return nil, err + } + type deleteSecurityGroupResponse struct { + XMLName xml.Name `xml:"DeleteSecurityGroupResponse"` + Return bool `xml:"return"` + } + return ec2XMLResponse(http.StatusOK, deleteSecurityGroupResponse{Return: true}) +} + +func (p *Provider) handleDescribeTags(_ url.Values) (*plugin.Response, error) { + tags, err := p.store.DescribeTags(defaultAccountID) + if err != nil { + return nil, err + } + type tagXML struct { + ResourceId string `xml:"resourceId"` + Key string `xml:"key"` + Value string `xml:"value"` + } + type describeTagsResponse struct { + XMLName xml.Name `xml:"DescribeTagsResponse"` + TagSet []tagXML `xml:"tagSet>item"` + } + items := make([]tagXML, 0, len(tags)) + for _, t := range tags { + items = append(items, tagXML{ResourceId: t.ResourceID, Key: t.Key, Value: t.Value}) + } + return ec2XMLResponse(http.StatusOK, describeTagsResponse{TagSet: items}) +} + +func (p *Provider) handleDeleteTags(form url.Values) (*plugin.Response, error) { + var resourceIDs []string + for i := 1; ; i++ { + id := form.Get(fmt.Sprintf("ResourceId.%d", i)) + if id == "" { + break + } + resourceIDs = append(resourceIDs, id) + } + var keys []string + for i := 1; ; i++ { + k := form.Get(fmt.Sprintf("Tag.%d.Key", i)) + if k == "" { + break + } + keys = append(keys, k) + } + if len(resourceIDs) == 0 { + return ec2XMLError("MissingParameter", "at least one ResourceId is required", http.StatusBadRequest), nil + } + if err := p.store.DeleteTags(defaultAccountID, resourceIDs, keys); err != nil { + return nil, err + } + type deleteTagsResponse struct { + XMLName xml.Name `xml:"DeleteTagsResponse"` + Return bool `xml:"return"` + } + return ec2XMLResponse(http.StatusOK, deleteTagsResponse{Return: true}) +} + +// sgRuleInput is one parsed security-group rule. +type sgRuleInput struct { + protocol string + fromPort int + toPort int + cidr string +} + +// parseSGRules extracts rules from either the structured IpPermissions.N.* form +// (what boto3 sends) or the legacy flat IpProtocol/FromPort/ToPort/CidrIp form. +func parseSGRules(form url.Values) []sgRuleInput { + var rules []sgRuleInput + for i := 1; ; i++ { + prefix := fmt.Sprintf("IpPermissions.%d.", i) + proto := form.Get(prefix + "IpProtocol") + if proto == "" { + break + } + from, _ := strconv.Atoi(form.Get(prefix + "FromPort")) + to, _ := strconv.Atoi(form.Get(prefix + "ToPort")) + matched := false + for j := 1; ; j++ { + cidr := form.Get(fmt.Sprintf("%sIpRanges.%d.CidrIp", prefix, j)) + if cidr == "" { + break + } + rules = append(rules, sgRuleInput{proto, from, to, cidr}) + matched = true + } + if !matched { + rules = append(rules, sgRuleInput{proto, from, to, ""}) + } + } + if len(rules) > 0 { + return rules + } + if proto := form.Get("IpProtocol"); proto != "" { + from, _ := strconv.Atoi(form.Get("FromPort")) + to, _ := strconv.Atoi(form.Get("ToPort")) + rules = append(rules, sgRuleInput{proto, from, to, form.Get("CidrIp")}) + } + return rules +} + +func (p *Provider) handleAuthorizeSecurityGroup(form url.Values, egress bool) (*plugin.Response, error) { + groupID := form.Get("GroupId") + if groupID == "" { + return ec2XMLError("MissingParameter", "GroupId is required", http.StatusBadRequest), nil + } + rules := parseSGRules(form) + if len(rules) == 0 { + return ec2XMLError("MissingParameter", "at least one rule is required", http.StatusBadRequest), nil + } + for _, r := range rules { + if _, err := p.store.AuthorizeSecurityGroupRule(defaultAccountID, groupID, egress, r.protocol, r.fromPort, r.toPort, r.cidr); err != nil { + if errors.Is(err, ErrSecurityGroupNotFound) { + return ec2XMLError("InvalidGroup.NotFound", err.Error(), http.StatusBadRequest), nil + } + return nil, err + } + } + respName := "AuthorizeSecurityGroupIngressResponse" + if egress { + respName = "AuthorizeSecurityGroupEgressResponse" + } + type authorizeResponse struct { + XMLName xml.Name + Return bool `xml:"return"` + } + return ec2XMLResponse(http.StatusOK, authorizeResponse{XMLName: xml.Name{Local: respName}, Return: true}) +} + +func (p *Provider) handleRevokeSecurityGroup(form url.Values, egress bool) (*plugin.Response, error) { + groupID := form.Get("GroupId") + if groupID == "" { + return ec2XMLError("MissingParameter", "GroupId is required", http.StatusBadRequest), nil + } + rules := parseSGRules(form) + if len(rules) == 0 { + return ec2XMLError("MissingParameter", "at least one rule is required", http.StatusBadRequest), nil + } + for _, r := range rules { + if err := p.store.RevokeSecurityGroupRule(defaultAccountID, groupID, egress, r.protocol, r.fromPort, r.toPort, r.cidr); err != nil { + return nil, err + } + } + respName := "RevokeSecurityGroupIngressResponse" + if egress { + respName = "RevokeSecurityGroupEgressResponse" + } + type revokeResponse struct { + XMLName xml.Name + Return bool `xml:"return"` + } + return ec2XMLResponse(http.StatusOK, revokeResponse{XMLName: xml.Name{Local: respName}, Return: true}) +} + // --- Volume handlers --- func (p *Provider) handleCreateVolume(form url.Values) (*plugin.Response, error) { diff --git a/internal/services/ec2/store.go b/internal/services/ec2/store.go index b861837d..67e8d86e 100644 --- a/internal/services/ec2/store.go +++ b/internal/services/ec2/store.go @@ -21,6 +21,7 @@ var ( ErrVpcNotFound = errors.New("vpc not found") ErrSubnetNotFound = errors.New("subnet not found") ErrSecurityGroupNotFound = errors.New("security group not found") + ErrAddressNotFound = errors.New("address not found") ErrVolumeNotFound = errors.New("volume not found") ErrSnapshotNotFound = errors.New("snapshot not found") ErrKeyPairNotFound = errors.New("key pair not found") @@ -178,6 +179,21 @@ var migrations = []sqlite.Migration{ ALTER TABLE vpcs ADD COLUMN enable_dns_support INTEGER NOT NULL DEFAULT 1; ALTER TABLE vpcs ADD COLUMN enable_dns_hostnames INTEGER NOT NULL DEFAULT 1; `}, + {Version: 8, SQL: ` + ALTER TABLE addresses ADD COLUMN association_id TEXT NOT NULL DEFAULT ''; + `}, + {Version: 9, SQL: ` + CREATE TABLE IF NOT EXISTS security_group_rules ( + rule_id TEXT PRIMARY KEY, + group_id TEXT NOT NULL, + egress INTEGER NOT NULL DEFAULT 0, + protocol TEXT NOT NULL DEFAULT '-1', + from_port INTEGER NOT NULL DEFAULT 0, + to_port INTEGER NOT NULL DEFAULT 0, + cidr TEXT NOT NULL DEFAULT '', + account_id TEXT NOT NULL + ); + `}, } type Instance struct { @@ -222,11 +238,23 @@ type Tag struct { } type Address struct { - AllocationID string - PublicIP string - Domain string - InstanceID string - AccountID string + AllocationID string + PublicIP string + Domain string + InstanceID string + AssociationID string + AccountID string +} + +type SecurityGroupRule struct { + RuleID string + GroupID string + Egress bool + Protocol string + FromPort int + ToPort int + CIDR string + AccountID string } type EC2Store struct { @@ -336,6 +364,42 @@ func (s *EC2Store) TerminateInstances(accountID string, ids []string) error { return nil } +// SetInstanceState sets the state of each instance and returns the previous +// state keyed by instance ID. Returns ErrInstanceNotFound if any id is unknown. +func (s *EC2Store) SetInstanceState(accountID string, ids []string, state string) (map[string]string, error) { + prev := make(map[string]string, len(ids)) + for _, id := range ids { + var cur string + err := s.db().QueryRow(`SELECT state FROM instances WHERE instance_id=? AND account_id=?`, id, accountID).Scan(&cur) + if errors.Is(err, sql.ErrNoRows) { + return nil, fmt.Errorf("%w: %s", ErrInstanceNotFound, id) + } + if err != nil { + return nil, err + } + if _, err := s.db().Exec(`UPDATE instances SET state=? WHERE instance_id=? AND account_id=?`, state, id, accountID); err != nil { + return nil, err + } + prev[id] = cur + } + return prev, nil +} + +// InstancesExist returns ErrInstanceNotFound if any id is unknown. +func (s *EC2Store) InstancesExist(accountID string, ids []string) error { + for _, id := range ids { + var one int + err := s.db().QueryRow(`SELECT 1 FROM instances WHERE instance_id=? AND account_id=?`, id, accountID).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("%w: %s", ErrInstanceNotFound, id) + } + if err != nil { + return err + } + } + return nil +} + // CreateVpc creates a new VPC. func (s *EC2Store) CreateVpc(accountID, cidrBlock string) (*Vpc, error) { id, err := randHex(8) @@ -505,6 +569,74 @@ func (s *EC2Store) GetTags(resourceID string) ([]Tag, error) { return out, rows.Err() } +// DescribeTags returns every tag for the account. +func (s *EC2Store) DescribeTags(accountID string) ([]Tag, error) { + rows, err := s.db().Query( + `SELECT resource_id, tag_key, tag_value FROM tags WHERE account_id=? ORDER BY resource_id, tag_key`, accountID, + ) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var out []Tag + for rows.Next() { + var t Tag + if err := rows.Scan(&t.ResourceID, &t.Key, &t.Value); err != nil { + return nil, err + } + out = append(out, t) + } + return out, rows.Err() +} + +// DeleteTags removes tags from the given resources. If keys is empty, all tags +// on each resource are removed. +func (s *EC2Store) DeleteTags(accountID string, resourceIDs, keys []string) error { + for _, resID := range resourceIDs { + if len(keys) == 0 { + if _, err := s.db().Exec(`DELETE FROM tags WHERE resource_id=? AND account_id=?`, resID, accountID); err != nil { + return err + } + continue + } + for _, k := range keys { + if _, err := s.db().Exec(`DELETE FROM tags WHERE resource_id=? AND tag_key=? AND account_id=?`, resID, k, accountID); err != nil { + return err + } + } + } + return nil +} + +// DeleteSecurityGroup removes a security group by group ID or group name. +func (s *EC2Store) DeleteSecurityGroup(accountID, identifier string) error { + res, err := s.db().Exec( + `DELETE FROM security_groups WHERE account_id=? AND (group_id=? OR group_name=?)`, + accountID, identifier, identifier, + ) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return ErrSecurityGroupNotFound + } + return nil +} + +// DeleteSubnet removes a subnet by ID. +func (s *EC2Store) DeleteSubnet(accountID, subnetID string) error { + res, err := s.db().Exec(`DELETE FROM subnets WHERE subnet_id=? AND account_id=?`, subnetID, accountID) + if err != nil { + return err + } + n, _ := res.RowsAffected() + if n == 0 { + return ErrSubnetNotFound + } + return nil +} + // AllocateAddress creates a new Elastic IP address. func (s *EC2Store) AllocateAddress(accountID, domain string) (*Address, error) { if domain == "" { @@ -529,6 +661,138 @@ func (s *EC2Store) AllocateAddress(accountID, domain string) (*Address, error) { return &Address{AllocationID: allocID, PublicIP: ip, Domain: domain, AccountID: accountID}, nil } +// DescribeAddresses returns all Elastic IP allocations for the account. +func (s *EC2Store) DescribeAddresses(accountID string) ([]Address, error) { + rows, err := s.db().Query( + `SELECT allocation_id, public_ip, domain, instance_id, association_id, account_id + FROM addresses WHERE account_id=? ORDER BY allocation_id`, accountID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var out []Address + for rows.Next() { + var a Address + if err := rows.Scan(&a.AllocationID, &a.PublicIP, &a.Domain, &a.InstanceID, &a.AssociationID, &a.AccountID); err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +// AssociateAddress binds an allocation to an instance and returns the new +// association ID. +func (s *EC2Store) AssociateAddress(accountID, allocationID, instanceID string) (string, error) { + id, err := randHex(8) + if err != nil { + return "", err + } + assocID := "eipassoc-" + id + res, err := s.db().Exec( + `UPDATE addresses SET instance_id=?, association_id=? WHERE allocation_id=? AND account_id=?`, + instanceID, assocID, allocationID, accountID) + if err != nil { + return "", err + } + if n, _ := res.RowsAffected(); n == 0 { + return "", ErrAddressNotFound + } + return assocID, nil +} + +// DisassociateAddress clears the association identified by associationID. +func (s *EC2Store) DisassociateAddress(accountID, associationID string) error { + res, err := s.db().Exec( + `UPDATE addresses SET instance_id='', association_id='' WHERE association_id=? AND account_id=?`, + associationID, accountID) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrAddressNotFound + } + return nil +} + +// ReleaseAddress removes an Elastic IP allocation. +func (s *EC2Store) ReleaseAddress(accountID, allocationID string) error { + res, err := s.db().Exec(`DELETE FROM addresses WHERE allocation_id=? AND account_id=?`, allocationID, accountID) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrAddressNotFound + } + return nil +} + +// AuthorizeSecurityGroupRule adds an ingress/egress rule to a security group +// and returns the new rule ID. Returns ErrSecurityGroupNotFound if the group +// does not exist. +func (s *EC2Store) AuthorizeSecurityGroupRule(accountID, groupID string, egress bool, protocol string, fromPort, toPort int, cidr string) (string, error) { + var one int + err := s.db().QueryRow(`SELECT 1 FROM security_groups WHERE group_id=? AND account_id=?`, groupID, accountID).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return "", ErrSecurityGroupNotFound + } + if err != nil { + return "", err + } + id, err := randHex(8) + if err != nil { + return "", err + } + ruleID := "sgr-" + id + egressInt := 0 + if egress { + egressInt = 1 + } + if _, err := s.db().Exec( + `INSERT INTO security_group_rules (rule_id, group_id, egress, protocol, from_port, to_port, cidr, account_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ruleID, groupID, egressInt, protocol, fromPort, toPort, cidr, accountID); err != nil { + return "", err + } + return ruleID, nil +} + +// RevokeSecurityGroupRule removes rules from a group matching the given +// direction and (protocol, fromPort, toPort, cidr) tuple. +func (s *EC2Store) RevokeSecurityGroupRule(accountID, groupID string, egress bool, protocol string, fromPort, toPort int, cidr string) error { + egressInt := 0 + if egress { + egressInt = 1 + } + _, err := s.db().Exec( + `DELETE FROM security_group_rules + WHERE group_id=? AND account_id=? AND egress=? AND protocol=? AND from_port=? AND to_port=? AND cidr=?`, + groupID, accountID, egressInt, protocol, fromPort, toPort, cidr) + return err +} + +// SecurityGroupRules returns the rules for a security group. +func (s *EC2Store) SecurityGroupRules(accountID, groupID string) ([]SecurityGroupRule, error) { + rows, err := s.db().Query( + `SELECT rule_id, group_id, egress, protocol, from_port, to_port, cidr, account_id + FROM security_group_rules WHERE group_id=? AND account_id=? ORDER BY rule_id`, groupID, accountID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var out []SecurityGroupRule + for rows.Next() { + var r SecurityGroupRule + var egressInt int + if err := rows.Scan(&r.RuleID, &r.GroupID, &egressInt, &r.Protocol, &r.FromPort, &r.ToPort, &r.CIDR, &r.AccountID); err != nil { + return nil, err + } + r.Egress = egressInt != 0 + out = append(out, r) + } + return out, rows.Err() +} + // ---- Volume operations ---- type Volume struct { diff --git a/internal/services/iam/provider.go b/internal/services/iam/provider.go index d4008e3e..8bd9a15b 100644 --- a/internal/services/iam/provider.go +++ b/internal/services/iam/provider.go @@ -179,6 +179,20 @@ func (p *IAMProvider) HandleRequest(ctx context.Context, op string, req *http.Re return p.handleAddUserToGroup(ctx, form) case "RemoveUserFromGroup": return p.handleRemoveUserFromGroup(ctx, form) + case "AttachGroupPolicy": + return p.handleAttachGroupPolicy(ctx, form) + case "DetachGroupPolicy": + return p.handleDetachGroupPolicy(ctx, form) + case "ListAttachedGroupPolicies": + return p.handleListAttachedGroupPolicies(ctx, form) + case "PutGroupPolicy": + return p.handlePutGroupPolicy(ctx, form) + case "GetGroupPolicy": + return p.handleGetGroupPolicy(ctx, form) + case "DeleteGroupPolicy": + return p.handleDeleteGroupPolicy(ctx, form) + case "ListGroupPolicies": + return p.handleListGroupPolicies(ctx, form) // Instance profiles case "CreateInstanceProfile": return p.handleCreateInstanceProfile(ctx, form) @@ -1033,6 +1047,51 @@ type listUserPoliciesResult struct { PolicyNames []string `xml:"PolicyNames>member"` } +type attachGroupPolicyResponse struct { + XMLName xml.Name `xml:"AttachGroupPolicyResponse"` +} + +type detachGroupPolicyResponse struct { + XMLName xml.Name `xml:"DetachGroupPolicyResponse"` +} + +type listAttachedGroupPoliciesResponse struct { + XMLName xml.Name `xml:"ListAttachedGroupPoliciesResponse"` + ListAttachedGroupPoliciesResult listAttachedGroupPoliciesResult `xml:"ListAttachedGroupPoliciesResult"` +} + +type listAttachedGroupPoliciesResult struct { + AttachedPolicies []attachedPolicyXML `xml:"AttachedPolicies>member"` +} + +type putGroupPolicyResponse struct { + XMLName xml.Name `xml:"PutGroupPolicyResponse"` +} + +type deleteGroupPolicyResponse struct { + XMLName xml.Name `xml:"DeleteGroupPolicyResponse"` +} + +type getGroupPolicyResponse struct { + XMLName xml.Name `xml:"GetGroupPolicyResponse"` + GetGroupPolicyResult getGroupPolicyResult `xml:"GetGroupPolicyResult"` +} + +type getGroupPolicyResult struct { + GroupName string `xml:"GroupName"` + PolicyName string `xml:"PolicyName"` + PolicyDocument string `xml:"PolicyDocument"` +} + +type listGroupPoliciesResponse struct { + XMLName xml.Name `xml:"ListGroupPoliciesResponse"` + ListGroupPoliciesResult listGroupPoliciesResult `xml:"ListGroupPoliciesResult"` +} + +type listGroupPoliciesResult struct { + PolicyNames []string `xml:"PolicyNames>member"` +} + type putRolePolicyResponse struct { XMLName xml.Name `xml:"PutRolePolicyResponse"` } @@ -1130,6 +1189,118 @@ func (p *IAMProvider) handleListUserPolicies(_ context.Context, form url.Values) }) } +func (p *IAMProvider) handleAttachGroupPolicy(_ context.Context, form url.Values) (*plugin.Response, error) { + groupName := form.Get("GroupName") + policyArn := form.Get("PolicyArn") + if groupName == "" || policyArn == "" { + return iamXMLError("MissingParameter", "GroupName and PolicyArn are required", http.StatusBadRequest), nil + } + if err := p.store.AttachGroupPolicy(defaultAccountID, groupName, policyArn); err != nil { + return nil, err + } + return iamXMLResponse(http.StatusOK, attachGroupPolicyResponse{}) +} + +func (p *IAMProvider) handleDetachGroupPolicy(_ context.Context, form url.Values) (*plugin.Response, error) { + groupName := form.Get("GroupName") + policyArn := form.Get("PolicyArn") + if groupName == "" || policyArn == "" { + return iamXMLError("MissingParameter", "GroupName and PolicyArn are required", http.StatusBadRequest), nil + } + if err := p.store.DetachGroupPolicy(defaultAccountID, groupName, policyArn); err != nil { + if err == ErrPolicyNotFound { + return iamXMLError("NoSuchEntity", fmt.Sprintf("The policy with ARN %s is not attached to group %s.", policyArn, groupName), http.StatusNotFound), nil + } + return nil, err + } + return iamXMLResponse(http.StatusOK, detachGroupPolicyResponse{}) +} + +func (p *IAMProvider) handleListAttachedGroupPolicies(_ context.Context, form url.Values) (*plugin.Response, error) { + groupName := form.Get("GroupName") + if groupName == "" { + return iamXMLError("MissingParameter", "GroupName is required", http.StatusBadRequest), nil + } + arns, err := p.store.ListAttachedGroupPolicies(defaultAccountID, groupName) + if err != nil { + return nil, err + } + members := make([]attachedPolicyXML, 0, len(arns)) + for _, arn := range arns { + name := arn + if i := strings.LastIndex(arn, "policy/"); i >= 0 { + name = arn[i+len("policy/"):] + } + members = append(members, attachedPolicyXML{PolicyArn: arn, PolicyName: name}) + } + return iamXMLResponse(http.StatusOK, listAttachedGroupPoliciesResponse{ + ListAttachedGroupPoliciesResult: listAttachedGroupPoliciesResult{AttachedPolicies: members}, + }) +} + +func (p *IAMProvider) handlePutGroupPolicy(_ context.Context, form url.Values) (*plugin.Response, error) { + groupName := form.Get("GroupName") + policyName := form.Get("PolicyName") + policyDocument := form.Get("PolicyDocument") + if groupName == "" || policyName == "" { + return iamXMLError("MissingParameter", "GroupName and PolicyName are required", http.StatusBadRequest), nil + } + if err := p.store.PutGroupInlinePolicy(defaultAccountID, groupName, policyName, policyDocument); err != nil { + return nil, err + } + return iamXMLResponse(http.StatusOK, putGroupPolicyResponse{}) +} + +func (p *IAMProvider) handleGetGroupPolicy(_ context.Context, form url.Values) (*plugin.Response, error) { + groupName := form.Get("GroupName") + policyName := form.Get("PolicyName") + if groupName == "" || policyName == "" { + return iamXMLError("MissingParameter", "GroupName and PolicyName are required", http.StatusBadRequest), nil + } + doc, err := p.store.GetGroupInlinePolicy(defaultAccountID, groupName, policyName) + if err != nil { + if err == ErrPolicyNotFound { + return iamXMLError("NoSuchEntity", fmt.Sprintf("The policy %s cannot be found.", policyName), http.StatusNotFound), nil + } + return nil, err + } + return iamXMLResponse(http.StatusOK, getGroupPolicyResponse{ + GetGroupPolicyResult: getGroupPolicyResult{GroupName: groupName, PolicyName: policyName, PolicyDocument: doc}, + }) +} + +func (p *IAMProvider) handleDeleteGroupPolicy(_ context.Context, form url.Values) (*plugin.Response, error) { + groupName := form.Get("GroupName") + policyName := form.Get("PolicyName") + if groupName == "" || policyName == "" { + return iamXMLError("MissingParameter", "GroupName and PolicyName are required", http.StatusBadRequest), nil + } + if err := p.store.DeleteGroupInlinePolicy(defaultAccountID, groupName, policyName); err != nil { + if err == ErrPolicyNotFound { + return iamXMLError("NoSuchEntity", fmt.Sprintf("The policy %s cannot be found.", policyName), http.StatusNotFound), nil + } + return nil, err + } + return iamXMLResponse(http.StatusOK, deleteGroupPolicyResponse{}) +} + +func (p *IAMProvider) handleListGroupPolicies(_ context.Context, form url.Values) (*plugin.Response, error) { + groupName := form.Get("GroupName") + if groupName == "" { + return iamXMLError("MissingParameter", "GroupName is required", http.StatusBadRequest), nil + } + names, err := p.store.ListGroupInlinePolicies(defaultAccountID, groupName) + if err != nil { + return nil, err + } + if names == nil { + names = []string{} + } + return iamXMLResponse(http.StatusOK, listGroupPoliciesResponse{ + ListGroupPoliciesResult: listGroupPoliciesResult{PolicyNames: names}, + }) +} + func (p *IAMProvider) handlePutRolePolicy(_ context.Context, form url.Values) (*plugin.Response, error) { roleName := form.Get("RoleName") policyName := form.Get("PolicyName") diff --git a/internal/services/iam/store.go b/internal/services/iam/store.go index 1bae57fe..8dc25a85 100644 --- a/internal/services/iam/store.go +++ b/internal/services/iam/store.go @@ -242,6 +242,22 @@ var iamMigrations = []sqlite.Migration{ PRIMARY KEY (role_name, tag_key, account_id) );`, }, + { + Version: 16, + SQL: `CREATE TABLE IF NOT EXISTS group_policies ( + group_name TEXT NOT NULL, + policy_arn TEXT NOT NULL, + account_id TEXT NOT NULL, + PRIMARY KEY (group_name, policy_arn, account_id) + ); + CREATE TABLE IF NOT EXISTS group_inline_policies ( + group_name TEXT NOT NULL, + policy_name TEXT NOT NULL, + policy_doc TEXT NOT NULL, + account_id TEXT NOT NULL, + PRIMARY KEY (group_name, policy_name, account_id) + );`, + }, } // IAMStore is a SQLite-backed store for IAM entities. @@ -1200,6 +1216,109 @@ func (s *IAMStore) ListUserInlinePolicies(accountID, userName string) ([]string, return names, rows.Err() } +// --- Group managed & inline policies (mirror the user/role variants) --- + +func (s *IAMStore) AttachGroupPolicy(accountID, groupName, policyArn string) error { + _, err := s.db().Exec( + `INSERT OR IGNORE INTO group_policies (group_name, policy_arn, account_id) VALUES (?, ?, ?);`, + groupName, policyArn, accountID, + ) + return err +} + +func (s *IAMStore) DetachGroupPolicy(accountID, groupName, policyArn string) error { + result, err := s.db().Exec( + `DELETE FROM group_policies WHERE group_name = ? AND policy_arn = ? AND account_id = ?;`, + groupName, policyArn, accountID, + ) + if err != nil { + return err + } + rows, _ := result.RowsAffected() + if rows == 0 { + return ErrPolicyNotFound + } + return nil +} + +func (s *IAMStore) ListAttachedGroupPolicies(accountID, groupName string) ([]string, error) { + rows, err := s.db().Query( + `SELECT policy_arn FROM group_policies WHERE group_name = ? AND account_id = ? ORDER BY policy_arn;`, + groupName, accountID, + ) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var arns []string + for rows.Next() { + var arn string + if err := rows.Scan(&arn); err != nil { + return nil, err + } + arns = append(arns, arn) + } + return arns, rows.Err() +} + +func (s *IAMStore) PutGroupInlinePolicy(accountID, groupName, policyName, policyDoc string) error { + _, err := s.db().Exec( + `INSERT OR REPLACE INTO group_inline_policies (group_name, policy_name, policy_doc, account_id) VALUES (?, ?, ?, ?);`, + groupName, policyName, policyDoc, accountID, + ) + return err +} + +func (s *IAMStore) GetGroupInlinePolicy(accountID, groupName, policyName string) (string, error) { + var doc string + err := s.db().QueryRow( + `SELECT policy_doc FROM group_inline_policies WHERE group_name = ? AND policy_name = ? AND account_id = ?;`, + groupName, policyName, accountID, + ).Scan(&doc) + if err != nil { + if err == sql.ErrNoRows { + return "", ErrPolicyNotFound + } + return "", err + } + return doc, nil +} + +func (s *IAMStore) DeleteGroupInlinePolicy(accountID, groupName, policyName string) error { + result, err := s.db().Exec( + `DELETE FROM group_inline_policies WHERE group_name = ? AND policy_name = ? AND account_id = ?;`, + groupName, policyName, accountID, + ) + if err != nil { + return err + } + rows, _ := result.RowsAffected() + if rows == 0 { + return ErrPolicyNotFound + } + return nil +} + +func (s *IAMStore) ListGroupInlinePolicies(accountID, groupName string) ([]string, error) { + rows, err := s.db().Query( + `SELECT policy_name FROM group_inline_policies WHERE group_name = ? AND account_id = ? ORDER BY policy_name;`, + groupName, accountID, + ) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var names []string + for rows.Next() { + var n string + if err := rows.Scan(&n); err != nil { + return nil, err + } + names = append(names, n) + } + return names, rows.Err() +} + // PutRoleInlinePolicy inserts or replaces an inline policy for a role. func (s *IAMStore) PutRoleInlinePolicy(accountID, roleName, policyName, policyDoc string) error { _, err := s.db().Exec( diff --git a/tests/compatibility/test_ec2.py b/tests/compatibility/test_ec2.py index 6a97d2bb..fb545b53 100644 --- a/tests/compatibility/test_ec2.py +++ b/tests/compatibility/test_ec2.py @@ -166,3 +166,142 @@ def test_network_acl(ec2_client): Egress=False, ) ec2_client.delete_network_acl(NetworkAclId=aclid) + + +def test_instance_stop_start(ec2_client): + iid = ec2_client.run_instances(ImageId="ami-life", MinCount=1, MaxCount=1)[ + "Instances" + ][0]["InstanceId"] + + stopped = ec2_client.stop_instances(InstanceIds=[iid]) + assert stopped["StoppingInstances"][0]["CurrentState"]["Name"] == "stopped" + assert stopped["StoppingInstances"][0]["PreviousState"]["Name"] == "running" + + started = ec2_client.start_instances(InstanceIds=[iid]) + assert started["StartingInstances"][0]["CurrentState"]["Name"] == "running" + assert started["StartingInstances"][0]["PreviousState"]["Name"] == "stopped" + + # Reboot succeeds without changing reported state. + ec2_client.reboot_instances(InstanceIds=[iid]) + + +def test_reboot_nonexistent_instance(ec2_client): + with pytest.raises(ClientError) as exc: + ec2_client.reboot_instances(InstanceIds=["i-doesnotexist0"]) + assert exc.value.response["Error"]["Code"] in ( + "InvalidInstanceID.NotFound", + "InvalidInstanceID.Malformed", + ) + + +def test_describe_availability_zones(ec2_client): + resp = ec2_client.describe_availability_zones() + zones = resp["AvailabilityZones"] + assert len(zones) >= 3 + assert any(z["ZoneName"] == "us-east-1a" for z in zones) + assert all(z["State"] == "available" for z in zones) + + +def test_describe_regions(ec2_client): + resp = ec2_client.describe_regions() + names = [r["RegionName"] for r in resp["Regions"]] + assert "us-east-1" in names + + +def test_delete_subnet(ec2_client): + vpc = ec2_client.create_vpc(CidrBlock="10.9.0.0/16") + subnet_id = ec2_client.create_subnet( + VpcId=vpc["Vpc"]["VpcId"], CidrBlock="10.9.1.0/24" + )["Subnet"]["SubnetId"] + ec2_client.delete_subnet(SubnetId=subnet_id) + remaining = [s["SubnetId"] for s in ec2_client.describe_subnets()["Subnets"]] + assert subnet_id not in remaining + + +def test_delete_security_group(ec2_client): + gid = ec2_client.create_security_group(GroupName="del-sg", Description="d")[ + "GroupId" + ] + ec2_client.delete_security_group(GroupId=gid) + ids = [ + sg["GroupId"] for sg in ec2_client.describe_security_groups()["SecurityGroups"] + ] + assert gid not in ids + + +def test_security_group_rules(ec2_client): + gid = ec2_client.create_security_group(GroupName="rules-sg", Description="d")[ + "GroupId" + ] + ec2_client.authorize_security_group_ingress( + GroupId=gid, + IpPermissions=[ + { + "IpProtocol": "tcp", + "FromPort": 22, + "ToPort": 22, + "IpRanges": [{"CidrIp": "0.0.0.0/0"}], + } + ], + ) + sg = [ + g + for g in ec2_client.describe_security_groups(GroupIds=[gid])["SecurityGroups"] + if g["GroupId"] == gid + ][0] + perms = sg["IpPermissions"] + assert any( + p["FromPort"] == 22 and p["IpRanges"][0]["CidrIp"] == "0.0.0.0/0" for p in perms + ) + + ec2_client.revoke_security_group_ingress( + GroupId=gid, + IpPermissions=[ + { + "IpProtocol": "tcp", + "FromPort": 22, + "ToPort": 22, + "IpRanges": [{"CidrIp": "0.0.0.0/0"}], + } + ], + ) + sg2 = [ + g + for g in ec2_client.describe_security_groups(GroupIds=[gid])["SecurityGroups"] + if g["GroupId"] == gid + ][0] + assert not sg2.get("IpPermissions") + + +def test_elastic_ip_association(ec2_client): + iid = ec2_client.run_instances(ImageId="ami-eip", MinCount=1, MaxCount=1)[ + "Instances" + ][0]["InstanceId"] + alloc = ec2_client.allocate_address(Domain="vpc") + alloc_id = alloc["AllocationId"] + + assoc = ec2_client.associate_address(AllocationId=alloc_id, InstanceId=iid) + assoc_id = assoc["AssociationId"] + assert assoc_id.startswith("eipassoc-") + + addrs = ec2_client.describe_addresses()["Addresses"] + mine = [a for a in addrs if a["AllocationId"] == alloc_id][0] + assert mine["InstanceId"] == iid + + ec2_client.disassociate_address(AssociationId=assoc_id) + ec2_client.release_address(AllocationId=alloc_id) + remaining = [ + a["AllocationId"] for a in ec2_client.describe_addresses()["Addresses"] + ] + assert alloc_id not in remaining + + +def test_delete_tags(ec2_client): + iid = ec2_client.run_instances(ImageId="ami-tagdel", MinCount=1, MaxCount=1)[ + "Instances" + ][0]["InstanceId"] + ec2_client.create_tags(Resources=[iid], Tags=[{"Key": "team", "Value": "x"}]) + ec2_client.delete_tags(Resources=[iid], Tags=[{"Key": "team"}]) + desc = ec2_client.describe_instances(InstanceIds=[iid]) + tags = desc["Reservations"][0]["Instances"][0].get("Tags", []) + assert "team" not in {t["Key"] for t in tags} diff --git a/tests/compatibility/test_iam_sts.py b/tests/compatibility/test_iam_sts.py index 436f384c..e787b026 100644 --- a/tests/compatibility/test_iam_sts.py +++ b/tests/compatibility/test_iam_sts.py @@ -282,3 +282,38 @@ def test_role_tags(iam_client): iam_client.untag_role(RoleName="tagrole", TagKeys=["purpose"]) resp2 = iam_client.list_role_tags(RoleName="tagrole") assert not any(t["Key"] == "purpose" for t in resp2["Tags"]) + + +def test_group_managed_policies(iam_client): + iam_client.create_group(GroupName="devs") + arn = "arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess" + iam_client.attach_group_policy(GroupName="devs", PolicyArn=arn) + attached = iam_client.list_attached_group_policies(GroupName="devs")[ + "AttachedPolicies" + ] + assert any(p["PolicyArn"] == arn for p in attached) + + iam_client.detach_group_policy(GroupName="devs", PolicyArn=arn) + attached2 = iam_client.list_attached_group_policies(GroupName="devs")[ + "AttachedPolicies" + ] + assert not any(p["PolicyArn"] == arn for p in attached2) + + +def test_group_inline_policies(iam_client): + iam_client.create_group(GroupName="ops") + doc = '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:*","Resource":"*"}]}' + iam_client.put_group_policy( + GroupName="ops", PolicyName="inline1", PolicyDocument=doc + ) + + names = iam_client.list_group_policies(GroupName="ops")["PolicyNames"] + assert "inline1" in names + + got = iam_client.get_group_policy(GroupName="ops", PolicyName="inline1") + assert got["GroupName"] == "ops" + assert got["PolicyName"] == "inline1" + + iam_client.delete_group_policy(GroupName="ops", PolicyName="inline1") + names2 = iam_client.list_group_policies(GroupName="ops")["PolicyNames"] + assert "inline1" not in names2 From 5330d518f59f1fb13f82cbd585cfbda9946046d6 Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Sat, 18 Jul 2026 23:30:49 +0900 Subject: [PATCH 2/4] docs: add changelog fragment for core-service op depth (#97) --- changes/unreleased/Added-20260718-233026.yaml | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 changes/unreleased/Added-20260718-233026.yaml diff --git a/changes/unreleased/Added-20260718-233026.yaml b/changes/unreleased/Added-20260718-233026.yaml new file mode 100644 index 00000000..a968c0ac --- /dev/null +++ b/changes/unreleased/Added-20260718-233026.yaml @@ -0,0 +1,5 @@ +kind: Added +body: 'Add 24 operations to core services: EC2 instance lifecycle (Start/Stop/Reboot), DescribeAvailabilityZones/DescribeRegions, security group rules (Authorize/Revoke ingress+egress), Elastic IP association, DeleteSecurityGroup/DeleteSubnet/DescribeTags/DeleteTags, and IAM group policy management (attach/detach/put/get/delete/list)' +time: 2026-07-18T23:30:26.666352+09:00 +custom: + Issue: "97" From f641079a4e9863cb521f2e318d8418c448ee9b54 Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Mon, 20 Jul 2026 06:01:37 +0900 Subject: [PATCH 3/4] fix(ec2): honor Describe filters/selectors and tighten resource validation Addresses a systemic pattern of Describe handlers discarding client input and store methods skipping existence/validation checks: - Describe{SecurityGroups,Tags,Addresses} now honor GroupId/AllocationId/ PublicIp selectors and Filter.N, and raise InvalidGroup.NotFound / InvalidAllocationID.NotFound for unknown IDs (were returning every resource in the account) - DeleteSecurityGroup resolves to a single group (group_name is not unique) and removes its rules, instead of deleting every same-named group - SetInstanceState validates all IDs first inside a transaction, so a bad ID in a batch Start/Stop no longer half-applies (EC2 is all-or-nothing) - AssociateAddress verifies the instance exists (InvalidInstanceID.NotFound) - Authorize rejects duplicate rules (InvalidPermission.Duplicate); Revoke reports unknown group/rule (InvalidGroup.NotFound / InvalidPermission.NotFound) - Authorize/Revoke accept GroupName, resolving the group once per request - All-traffic ('-1') rules omit FromPort/ToPort instead of reporting 0 - DeleteTags honors Tag.N.Value (value-qualified delete) - DescribeSecurityGroups fetches rules in one batched query (was N+1) Adds 12 boto3 regression tests covering each behavior. --- changes/unreleased/Fixed-20260720-054448.yaml | 5 + internal/services/ec2/provider.go | 264 ++++++++++++++---- internal/services/ec2/store.go | 209 ++++++++++---- tests/compatibility/test_ec2.py | 174 ++++++++++++ 4 files changed, 529 insertions(+), 123 deletions(-) create mode 100644 changes/unreleased/Fixed-20260720-054448.yaml diff --git a/changes/unreleased/Fixed-20260720-054448.yaml b/changes/unreleased/Fixed-20260720-054448.yaml new file mode 100644 index 00000000..44fc386f --- /dev/null +++ b/changes/unreleased/Fixed-20260720-054448.yaml @@ -0,0 +1,5 @@ +kind: Fixed +body: 'EC2: Describe{SecurityGroups,Tags,Addresses} now honor GroupId/AllocationId/PublicIp selectors and Filters (and raise NotFound for unknown IDs) instead of returning every resource; delete-by-name removes a single security group; batch Start/Stop is atomic; AssociateAddress validates the instance; duplicate/absent security-group rules are rejected; Authorize/Revoke accept GroupName; all-traffic rules omit ports' +time: 2026-07-20T05:44:48.649862+09:00 +custom: + Issue: "97" diff --git a/internal/services/ec2/provider.go b/internal/services/ec2/provider.go index b5f467e0..ca17a9c9 100644 --- a/internal/services/ec2/provider.go +++ b/internal/services/ec2/provider.go @@ -296,14 +296,7 @@ func (p *Provider) handleRunInstances(form url.Values) (*plugin.Response, error) } func (p *Provider) handleDescribeInstances(form url.Values) (*plugin.Response, error) { - var ids []string - for i := 1; ; i++ { - id := form.Get(fmt.Sprintf("InstanceId.%d", i)) - if id == "" { - break - } - ids = append(ids, id) - } + ids := collectIndexed(form, "InstanceId") instances, err := p.store.DescribeInstances(defaultAccountID, ids) if err != nil { @@ -357,14 +350,7 @@ func (p *Provider) handleDescribeInstances(form url.Values) (*plugin.Response, e } func (p *Provider) handleTerminateInstances(form url.Values) (*plugin.Response, error) { - var ids []string - for i := 1; ; i++ { - id := form.Get(fmt.Sprintf("InstanceId.%d", i)) - if id == "" { - break - } - ids = append(ids, id) - } + ids := collectIndexed(form, "InstanceId") if len(ids) == 0 { return ec2XMLError("MissingParameter", "at least one InstanceId is required", http.StatusBadRequest), nil } @@ -399,23 +385,69 @@ func (p *Provider) handleTerminateInstances(form url.Values) (*plugin.Response, return ec2XMLResponse(http.StatusOK, terminateInstancesResponse{InstancesSet: items}) } -// instanceIDsFromForm collects InstanceId.1, InstanceId.2, … from a Query form. -func instanceIDsFromForm(form url.Values) []string { - var ids []string +// collectIndexed collects the EC2 list convention prefix.1, prefix.2, … from a +// Query form (e.g. InstanceId.1, GroupId.1, Filter.3.Value.2). +func collectIndexed(form url.Values, prefix string) []string { + var out []string for i := 1; ; i++ { - id := form.Get(fmt.Sprintf("InstanceId.%d", i)) - if id == "" { + v := form.Get(fmt.Sprintf("%s.%d", prefix, i)) + if v == "" { break } - ids = append(ids, id) + out = append(out, v) + } + return out +} + +func containsStr(s []string, v string) bool { + for _, x := range s { + if x == v { + return true + } + } + return false +} + +// ec2Filter is a parsed Filter.N.Name / Filter.N.Value.M entry. +type ec2Filter struct { + name string + values []string +} + +func parseFilters(form url.Values) []ec2Filter { + var out []ec2Filter + for i := 1; ; i++ { + name := form.Get(fmt.Sprintf("Filter.%d.Name", i)) + if name == "" { + break + } + out = append(out, ec2Filter{name: name, values: collectIndexed(form, fmt.Sprintf("Filter.%d.Value", i))}) } - return ids + return out +} + +// matchFilters reports whether attrs satisfy every filter (AND across filters, +// OR within a filter's values). +// ponytail: accepted — filter names absent from attrs are ignored rather than +// erroring. Deliberate for a mock: erroring would reject valid unmodeled filters +// (tag:*, resource-type, instance-state-name, …). Revisit only if a test needs strictness. +func matchFilters(filters []ec2Filter, attrs map[string]string) bool { + for _, f := range filters { + v, ok := attrs[f.name] + if !ok { + continue + } + if !containsStr(f.values, v) { + return false + } + } + return true } // changeInstanceState handles StartInstances/StopInstances: it sets the new // state and reports the per-instance previous → current transition. func (p *Provider) changeInstanceState(form url.Values, respName, newState string) (*plugin.Response, error) { - ids := instanceIDsFromForm(form) + ids := collectIndexed(form, "InstanceId") if len(ids) == 0 { return ec2XMLError("MissingParameter", "at least one InstanceId is required", http.StatusBadRequest), nil } @@ -452,7 +484,7 @@ func (p *Provider) changeInstanceState(form url.Values, respName, newState strin // handleRebootInstances validates the instances exist; reboot does not change // the reported state (mirrors the real EC2 behaviour). func (p *Provider) handleRebootInstances(form url.Values) (*plugin.Response, error) { - ids := instanceIDsFromForm(form) + ids := collectIndexed(form, "InstanceId") if len(ids) == 0 { return ec2XMLError("MissingParameter", "at least one InstanceId is required", http.StatusBadRequest), nil } @@ -594,11 +626,36 @@ func (p *Provider) handleCreateSecurityGroup(form url.Values) (*plugin.Response, return ec2XMLResponse(http.StatusOK, createSecurityGroupResponse{GroupId: sg.GroupID}) } -func (p *Provider) handleDescribeSecurityGroups(_ url.Values) (*plugin.Response, error) { +func (p *Provider) handleDescribeSecurityGroups(form url.Values) (*plugin.Response, error) { sgs, err := p.store.DescribeSecurityGroups(defaultAccountID) if err != nil { return nil, err } + rulesByGroup, err := p.store.AllSecurityGroupRules(defaultAccountID) + if err != nil { + return nil, err + } + + wantIDs := collectIndexed(form, "GroupId") + wantNames := collectIndexed(form, "GroupName") + filters := parseFilters(form) + + // A requested GroupId/GroupName that matches no existing group is an error. + idSet, nameSet := map[string]bool{}, map[string]bool{} + for _, sg := range sgs { + idSet[sg.GroupID], nameSet[sg.GroupName] = true, true + } + for _, id := range wantIDs { + if !idSet[id] { + return ec2XMLError("InvalidGroup.NotFound", fmt.Sprintf("The security group '%s' does not exist", id), http.StatusBadRequest), nil + } + } + for _, n := range wantNames { + if !nameSet[n] { + return ec2XMLError("InvalidGroup.NotFound", fmt.Sprintf("The security group '%s' does not exist", n), http.StatusBadRequest), nil + } + } + type ipRangeXML struct { CidrIp string `xml:"cidrIp"` } @@ -621,8 +678,7 @@ func (p *Provider) handleDescribeSecurityGroups(_ url.Values) (*plugin.Response, SecurityGroupInfo []sgItemXML `xml:"securityGroupInfo>item"` } toPerm := func(r SecurityGroupRule) ipPermissionXML { - from, to := r.FromPort, r.ToPort - perm := ipPermissionXML{IpProtocol: r.Protocol, FromPort: &from, ToPort: &to} + perm := ipPermissionXML{IpProtocol: r.Protocol, FromPort: r.FromPort, ToPort: r.ToPort} if r.CIDR != "" { perm.IpRanges = []ipRangeXML{{CidrIp: r.CIDR}} } @@ -630,12 +686,19 @@ func (p *Provider) handleDescribeSecurityGroups(_ url.Values) (*plugin.Response, } items := make([]sgItemXML, 0, len(sgs)) for _, sg := range sgs { - item := sgItemXML{GroupId: sg.GroupID, GroupName: sg.GroupName, VpcId: sg.VpcID, Description: sg.Description} - rules, err := p.store.SecurityGroupRules(defaultAccountID, sg.GroupID) - if err != nil { - return nil, err + if (len(wantIDs) > 0 || len(wantNames) > 0) && !containsStr(wantIDs, sg.GroupID) && !containsStr(wantNames, sg.GroupName) { + continue + } + if !matchFilters(filters, map[string]string{ + "group-id": sg.GroupID, + "group-name": sg.GroupName, + "vpc-id": sg.VpcID, + "description": sg.Description, + }) { + continue } - for _, r := range rules { + item := sgItemXML{GroupId: sg.GroupID, GroupName: sg.GroupName, VpcId: sg.VpcID, Description: sg.Description} + for _, r := range rulesByGroup[sg.GroupID] { if r.Egress { item.IpPermissionsEgress = append(item.IpPermissionsEgress, toPerm(r)) } else { @@ -705,6 +768,9 @@ func (p *Provider) handleAssociateAddress(form url.Values) (*plugin.Response, er } assocID, err := p.store.AssociateAddress(defaultAccountID, allocationID, instanceID) if err != nil { + if errors.Is(err, ErrInstanceNotFound) { + return ec2XMLError("InvalidInstanceID.NotFound", err.Error(), http.StatusBadRequest), nil + } if errors.Is(err, ErrAddressNotFound) { return ec2XMLError("InvalidAllocationID.NotFound", err.Error(), http.StatusBadRequest), nil } @@ -754,11 +820,32 @@ func (p *Provider) handleReleaseAddress(form url.Values) (*plugin.Response, erro return ec2XMLResponse(http.StatusOK, releaseAddressResponse{Return: true}) } -func (p *Provider) handleDescribeAddresses(_ url.Values) (*plugin.Response, error) { +func (p *Provider) handleDescribeAddresses(form url.Values) (*plugin.Response, error) { addrs, err := p.store.DescribeAddresses(defaultAccountID) if err != nil { return nil, err } + + wantAllocs := collectIndexed(form, "AllocationId") + wantIPs := collectIndexed(form, "PublicIp") + filters := parseFilters(form) + + // A requested AllocationId/PublicIp that matches nothing is an error in EC2. + allocSet, ipSet := map[string]bool{}, map[string]bool{} + for _, a := range addrs { + allocSet[a.AllocationID], ipSet[a.PublicIP] = true, true + } + for _, id := range wantAllocs { + if !allocSet[id] { + return ec2XMLError("InvalidAllocationID.NotFound", fmt.Sprintf("Address with allocation ID '%s' not found", id), http.StatusBadRequest), nil + } + } + for _, ip := range wantIPs { + if !ipSet[ip] { + return ec2XMLError("InvalidAddress.NotFound", fmt.Sprintf("Address '%s' not found", ip), http.StatusBadRequest), nil + } + } + type addrXML struct { PublicIp string `xml:"publicIp"` AllocationId string `xml:"allocationId"` @@ -772,6 +859,17 @@ func (p *Provider) handleDescribeAddresses(_ url.Values) (*plugin.Response, erro } items := make([]addrXML, 0, len(addrs)) for _, a := range addrs { + if (len(wantAllocs) > 0 || len(wantIPs) > 0) && !containsStr(wantAllocs, a.AllocationID) && !containsStr(wantIPs, a.PublicIP) { + continue + } + if !matchFilters(filters, map[string]string{ + "allocation-id": a.AllocationID, + "public-ip": a.PublicIP, + "instance-id": a.InstanceID, + "domain": a.Domain, + }) { + continue + } items = append(items, addrXML{ PublicIp: a.PublicIP, AllocationId: a.AllocationID, Domain: a.Domain, InstanceId: a.InstanceID, AssociationId: a.AssociationID, @@ -862,11 +960,12 @@ func (p *Provider) handleDeleteSecurityGroup(form url.Values) (*plugin.Response, return ec2XMLResponse(http.StatusOK, deleteSecurityGroupResponse{Return: true}) } -func (p *Provider) handleDescribeTags(_ url.Values) (*plugin.Response, error) { +func (p *Provider) handleDescribeTags(form url.Values) (*plugin.Response, error) { tags, err := p.store.DescribeTags(defaultAccountID) if err != nil { return nil, err } + filters := parseFilters(form) type tagXML struct { ResourceId string `xml:"resourceId"` Key string `xml:"key"` @@ -878,32 +977,32 @@ func (p *Provider) handleDescribeTags(_ url.Values) (*plugin.Response, error) { } items := make([]tagXML, 0, len(tags)) for _, t := range tags { + if !matchFilters(filters, map[string]string{ + "resource-id": t.ResourceID, + "key": t.Key, + "value": t.Value, + }) { + continue + } items = append(items, tagXML{ResourceId: t.ResourceID, Key: t.Key, Value: t.Value}) } return ec2XMLResponse(http.StatusOK, describeTagsResponse{TagSet: items}) } func (p *Provider) handleDeleteTags(form url.Values) (*plugin.Response, error) { - var resourceIDs []string - for i := 1; ; i++ { - id := form.Get(fmt.Sprintf("ResourceId.%d", i)) - if id == "" { - break - } - resourceIDs = append(resourceIDs, id) - } - var keys []string + resourceIDs := collectIndexed(form, "ResourceId") + var tags []Tag for i := 1; ; i++ { k := form.Get(fmt.Sprintf("Tag.%d.Key", i)) if k == "" { break } - keys = append(keys, k) + tags = append(tags, Tag{Key: k, Value: form.Get(fmt.Sprintf("Tag.%d.Value", i))}) } if len(resourceIDs) == 0 { return ec2XMLError("MissingParameter", "at least one ResourceId is required", http.StatusBadRequest), nil } - if err := p.store.DeleteTags(defaultAccountID, resourceIDs, keys); err != nil { + if err := p.store.DeleteTags(defaultAccountID, resourceIDs, tags); err != nil { return nil, err } type deleteTagsResponse struct { @@ -913,14 +1012,27 @@ func (p *Provider) handleDeleteTags(form url.Values) (*plugin.Response, error) { return ec2XMLResponse(http.StatusOK, deleteTagsResponse{Return: true}) } -// sgRuleInput is one parsed security-group rule. +// sgRuleInput is one parsed security-group rule. fromPort/toPort are nil for +// all-port rules (e.g. protocol "-1"), which AWS omits rather than reporting 0. type sgRuleInput struct { protocol string - fromPort int - toPort int + fromPort *int + toPort *int cidr string } +// atoiPtr parses an optional port: absent or unparseable → nil (all ports). +func atoiPtr(s string) *int { + if s == "" { + return nil + } + n, err := strconv.Atoi(s) + if err != nil { + return nil + } + return &n +} + // parseSGRules extracts rules from either the structured IpPermissions.N.* form // (what boto3 sends) or the legacy flat IpProtocol/FromPort/ToPort/CidrIp form. func parseSGRules(form url.Values) []sgRuleInput { @@ -931,8 +1043,8 @@ func parseSGRules(form url.Values) []sgRuleInput { if proto == "" { break } - from, _ := strconv.Atoi(form.Get(prefix + "FromPort")) - to, _ := strconv.Atoi(form.Get(prefix + "ToPort")) + from := atoiPtr(form.Get(prefix + "FromPort")) + to := atoiPtr(form.Get(prefix + "ToPort")) matched := false for j := 1; ; j++ { cidr := form.Get(fmt.Sprintf("%sIpRanges.%d.CidrIp", prefix, j)) @@ -950,17 +1062,39 @@ func parseSGRules(form url.Values) []sgRuleInput { return rules } if proto := form.Get("IpProtocol"); proto != "" { - from, _ := strconv.Atoi(form.Get("FromPort")) - to, _ := strconv.Atoi(form.Get("ToPort")) - rules = append(rules, sgRuleInput{proto, from, to, form.Get("CidrIp")}) + rules = append(rules, sgRuleInput{proto, atoiPtr(form.Get("FromPort")), atoiPtr(form.Get("ToPort")), form.Get("CidrIp")}) } return rules } +// resolveSGForRuleChange reads GroupId or GroupName from the form and resolves +// it to a group_id once for the whole request. It returns an error +// *plugin.Response when the parameter is missing or the group does not exist. +func (p *Provider) resolveSGForRuleChange(form url.Values) (string, *plugin.Response, error) { + id := form.Get("GroupId") + if id == "" { + id = form.Get("GroupName") + } + if id == "" { + return "", ec2XMLError("MissingParameter", "GroupId or GroupName is required", http.StatusBadRequest), nil + } + groupID, err := p.store.ResolveSecurityGroupID(defaultAccountID, id) + if err != nil { + if errors.Is(err, ErrSecurityGroupNotFound) { + return "", ec2XMLError("InvalidGroup.NotFound", err.Error(), http.StatusBadRequest), nil + } + return "", nil, err + } + return groupID, nil, nil +} + func (p *Provider) handleAuthorizeSecurityGroup(form url.Values, egress bool) (*plugin.Response, error) { - groupID := form.Get("GroupId") - if groupID == "" { - return ec2XMLError("MissingParameter", "GroupId is required", http.StatusBadRequest), nil + groupID, resp, err := p.resolveSGForRuleChange(form) + if err != nil { + return nil, err + } + if resp != nil { + return resp, nil } rules := parseSGRules(form) if len(rules) == 0 { @@ -968,8 +1102,8 @@ func (p *Provider) handleAuthorizeSecurityGroup(form url.Values, egress bool) (* } for _, r := range rules { if _, err := p.store.AuthorizeSecurityGroupRule(defaultAccountID, groupID, egress, r.protocol, r.fromPort, r.toPort, r.cidr); err != nil { - if errors.Is(err, ErrSecurityGroupNotFound) { - return ec2XMLError("InvalidGroup.NotFound", err.Error(), http.StatusBadRequest), nil + if errors.Is(err, ErrSecurityGroupRuleExists) { + return ec2XMLError("InvalidPermission.Duplicate", err.Error(), http.StatusBadRequest), nil } return nil, err } @@ -986,9 +1120,12 @@ func (p *Provider) handleAuthorizeSecurityGroup(form url.Values, egress bool) (* } func (p *Provider) handleRevokeSecurityGroup(form url.Values, egress bool) (*plugin.Response, error) { - groupID := form.Get("GroupId") - if groupID == "" { - return ec2XMLError("MissingParameter", "GroupId is required", http.StatusBadRequest), nil + groupID, resp, err := p.resolveSGForRuleChange(form) + if err != nil { + return nil, err + } + if resp != nil { + return resp, nil } rules := parseSGRules(form) if len(rules) == 0 { @@ -996,6 +1133,9 @@ func (p *Provider) handleRevokeSecurityGroup(form url.Values, egress bool) (*plu } for _, r := range rules { if err := p.store.RevokeSecurityGroupRule(defaultAccountID, groupID, egress, r.protocol, r.fromPort, r.toPort, r.cidr); err != nil { + if errors.Is(err, ErrSecurityGroupRuleNotFound) { + return ec2XMLError("InvalidPermission.NotFound", err.Error(), http.StatusBadRequest), nil + } return nil, err } } diff --git a/internal/services/ec2/store.go b/internal/services/ec2/store.go index 67e8d86e..3a939d7b 100644 --- a/internal/services/ec2/store.go +++ b/internal/services/ec2/store.go @@ -17,19 +17,21 @@ import ( ) var ( - ErrInstanceNotFound = errors.New("instance not found") - ErrVpcNotFound = errors.New("vpc not found") - ErrSubnetNotFound = errors.New("subnet not found") - ErrSecurityGroupNotFound = errors.New("security group not found") - ErrAddressNotFound = errors.New("address not found") - ErrVolumeNotFound = errors.New("volume not found") - ErrSnapshotNotFound = errors.New("snapshot not found") - ErrKeyPairNotFound = errors.New("key pair not found") - ErrKeyPairAlreadyExists = errors.New("key pair already exists") - ErrImageNotFound = errors.New("image not found") - ErrRouteTableNotFound = errors.New("route table not found") - ErrInternetGatewayNotFound = errors.New("internet gateway not found") - ErrNetworkACLNotFound = errors.New("network acl not found") + ErrInstanceNotFound = errors.New("instance not found") + ErrVpcNotFound = errors.New("vpc not found") + ErrSubnetNotFound = errors.New("subnet not found") + ErrSecurityGroupNotFound = errors.New("security group not found") + ErrSecurityGroupRuleNotFound = errors.New("security group rule not found") + ErrSecurityGroupRuleExists = errors.New("security group rule already exists") + ErrAddressNotFound = errors.New("address not found") + ErrVolumeNotFound = errors.New("volume not found") + ErrSnapshotNotFound = errors.New("snapshot not found") + ErrKeyPairNotFound = errors.New("key pair not found") + ErrKeyPairAlreadyExists = errors.New("key pair already exists") + ErrImageNotFound = errors.New("image not found") + ErrRouteTableNotFound = errors.New("route table not found") + ErrInternetGatewayNotFound = errors.New("internet gateway not found") + ErrNetworkACLNotFound = errors.New("network acl not found") ) var migrations = []sqlite.Migration{ @@ -188,8 +190,8 @@ var migrations = []sqlite.Migration{ group_id TEXT NOT NULL, egress INTEGER NOT NULL DEFAULT 0, protocol TEXT NOT NULL DEFAULT '-1', - from_port INTEGER NOT NULL DEFAULT 0, - to_port INTEGER NOT NULL DEFAULT 0, + from_port INTEGER, + to_port INTEGER, cidr TEXT NOT NULL DEFAULT '', account_id TEXT NOT NULL ); @@ -251,8 +253,8 @@ type SecurityGroupRule struct { GroupID string Egress bool Protocol string - FromPort int - ToPort int + FromPort *int // nil when the rule spans all ports (e.g. protocol "-1") + ToPort *int CIDR string AccountID string } @@ -367,20 +369,33 @@ func (s *EC2Store) TerminateInstances(accountID string, ids []string) error { // SetInstanceState sets the state of each instance and returns the previous // state keyed by instance ID. Returns ErrInstanceNotFound if any id is unknown. func (s *EC2Store) SetInstanceState(accountID string, ids []string, state string) (map[string]string, error) { + tx, err := s.db().Begin() + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback() }() + + // Validate every id (and capture its previous state) before mutating any, + // so a bad id in the batch leaves all instances untouched (EC2 is all-or-nothing). prev := make(map[string]string, len(ids)) for _, id := range ids { var cur string - err := s.db().QueryRow(`SELECT state FROM instances WHERE instance_id=? AND account_id=?`, id, accountID).Scan(&cur) + err := tx.QueryRow(`SELECT state FROM instances WHERE instance_id=? AND account_id=?`, id, accountID).Scan(&cur) if errors.Is(err, sql.ErrNoRows) { return nil, fmt.Errorf("%w: %s", ErrInstanceNotFound, id) } if err != nil { return nil, err } - if _, err := s.db().Exec(`UPDATE instances SET state=? WHERE instance_id=? AND account_id=?`, state, id, accountID); err != nil { + prev[id] = cur + } + for _, id := range ids { + if _, err := tx.Exec(`UPDATE instances SET state=? WHERE instance_id=? AND account_id=?`, state, id, accountID); err != nil { return nil, err } - prev[id] = cur + } + if err := tx.Commit(); err != nil { + return nil, err } return prev, nil } @@ -589,18 +604,25 @@ func (s *EC2Store) DescribeTags(accountID string) ([]Tag, error) { return out, rows.Err() } -// DeleteTags removes tags from the given resources. If keys is empty, all tags -// on each resource are removed. -func (s *EC2Store) DeleteTags(accountID string, resourceIDs, keys []string) error { +// DeleteTags removes tags from the given resources. If tags is empty, all tags +// on each resource are removed. A tag whose Value is set deletes only the +// matching (key, value) pair; a tag with an empty Value deletes by key alone. +func (s *EC2Store) DeleteTags(accountID string, resourceIDs []string, tags []Tag) error { for _, resID := range resourceIDs { - if len(keys) == 0 { + if len(tags) == 0 { if _, err := s.db().Exec(`DELETE FROM tags WHERE resource_id=? AND account_id=?`, resID, accountID); err != nil { return err } continue } - for _, k := range keys { - if _, err := s.db().Exec(`DELETE FROM tags WHERE resource_id=? AND tag_key=? AND account_id=?`, resID, k, accountID); err != nil { + for _, t := range tags { + var err error + if t.Value != "" { + _, err = s.db().Exec(`DELETE FROM tags WHERE resource_id=? AND tag_key=? AND tag_value=? AND account_id=?`, resID, t.Key, t.Value, accountID) + } else { + _, err = s.db().Exec(`DELETE FROM tags WHERE resource_id=? AND tag_key=? AND account_id=?`, resID, t.Key, accountID) + } + if err != nil { return err } } @@ -608,19 +630,36 @@ func (s *EC2Store) DeleteTags(accountID string, resourceIDs, keys []string) erro return nil } -// DeleteSecurityGroup removes a security group by group ID or group name. -func (s *EC2Store) DeleteSecurityGroup(accountID, identifier string) error { - res, err := s.db().Exec( - `DELETE FROM security_groups WHERE account_id=? AND (group_id=? OR group_name=?)`, +// ResolveSecurityGroupID returns the group_id for an identifier that may be a +// group ID or a group name. group_name is not unique, so a name that matches +// several groups resolves to the lowest group_id. Returns +// ErrSecurityGroupNotFound if nothing matches. +func (s *EC2Store) ResolveSecurityGroupID(accountID, identifier string) (string, error) { + var groupID string + err := s.db().QueryRow( + `SELECT group_id FROM security_groups WHERE account_id=? AND (group_id=? OR group_name=?) ORDER BY group_id LIMIT 1`, accountID, identifier, identifier, - ) + ).Scan(&groupID) + if errors.Is(err, sql.ErrNoRows) { + return "", ErrSecurityGroupNotFound + } + if err != nil { + return "", err + } + return groupID, nil +} + +// DeleteSecurityGroup removes a single security group by group ID or group name, +// along with its rules. Deleting by a non-unique name removes exactly one group. +func (s *EC2Store) DeleteSecurityGroup(accountID, identifier string) error { + groupID, err := s.ResolveSecurityGroupID(accountID, identifier) if err != nil { return err } - n, _ := res.RowsAffected() - if n == 0 { - return ErrSecurityGroupNotFound + if _, err := s.db().Exec(`DELETE FROM security_groups WHERE group_id=? AND account_id=?`, groupID, accountID); err != nil { + return err } + _, _ = s.db().Exec(`DELETE FROM security_group_rules WHERE group_id=? AND account_id=?`, groupID, accountID) return nil } @@ -684,6 +723,14 @@ func (s *EC2Store) DescribeAddresses(accountID string) ([]Address, error) { // AssociateAddress binds an allocation to an instance and returns the new // association ID. func (s *EC2Store) AssociateAddress(accountID, allocationID, instanceID string) (string, error) { + var one int + err := s.db().QueryRow(`SELECT 1 FROM instances WHERE instance_id=? AND account_id=?`, instanceID, accountID).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return "", ErrInstanceNotFound + } + if err != nil { + return "", err + } id, err := randHex(8) if err != nil { return "", err @@ -728,15 +775,24 @@ func (s *EC2Store) ReleaseAddress(accountID, allocationID string) error { } // AuthorizeSecurityGroupRule adds an ingress/egress rule to a security group -// and returns the new rule ID. Returns ErrSecurityGroupNotFound if the group -// does not exist. -func (s *EC2Store) AuthorizeSecurityGroupRule(accountID, groupID string, egress bool, protocol string, fromPort, toPort int, cidr string) (string, error) { - var one int - err := s.db().QueryRow(`SELECT 1 FROM security_groups WHERE group_id=? AND account_id=?`, groupID, accountID).Scan(&one) - if errors.Is(err, sql.ErrNoRows) { - return "", ErrSecurityGroupNotFound +// and returns the new rule ID. The caller must ensure the group exists (see +// ResolveSecurityGroupID). Returns ErrSecurityGroupRuleExists if an identical +// rule is already present. fromPort/toPort are nil for all-port rules. +func (s *EC2Store) AuthorizeSecurityGroupRule(accountID, groupID string, egress bool, protocol string, fromPort, toPort *int, cidr string) (string, error) { + egressInt := 0 + if egress { + egressInt = 1 } - if err != nil { + // SQLite IS compares like = but treats NULLs as equal, so nil ports match nil. + var dup int + err := s.db().QueryRow( + `SELECT 1 FROM security_group_rules + WHERE group_id=? AND account_id=? AND egress=? AND protocol=? AND from_port IS ? AND to_port IS ? AND cidr=?`, + groupID, accountID, egressInt, protocol, fromPort, toPort, cidr).Scan(&dup) + if err == nil { + return "", ErrSecurityGroupRuleExists + } + if !errors.Is(err, sql.ErrNoRows) { return "", err } id, err := randHex(8) @@ -744,10 +800,6 @@ func (s *EC2Store) AuthorizeSecurityGroupRule(accountID, groupID string, egress return "", err } ruleID := "sgr-" + id - egressInt := 0 - if egress { - egressInt = 1 - } if _, err := s.db().Exec( `INSERT INTO security_group_rules (rule_id, group_id, egress, protocol, from_port, to_port, cidr, account_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, @@ -757,42 +809,77 @@ func (s *EC2Store) AuthorizeSecurityGroupRule(accountID, groupID string, egress return ruleID, nil } -// RevokeSecurityGroupRule removes rules from a group matching the given -// direction and (protocol, fromPort, toPort, cidr) tuple. -func (s *EC2Store) RevokeSecurityGroupRule(accountID, groupID string, egress bool, protocol string, fromPort, toPort int, cidr string) error { +// RevokeSecurityGroupRule removes the rule matching the given direction and +// (protocol, fromPort, toPort, cidr) tuple. Returns ErrSecurityGroupNotFound if +// the group is unknown and ErrSecurityGroupRuleNotFound if no such rule exists. +func (s *EC2Store) RevokeSecurityGroupRule(accountID, groupID string, egress bool, protocol string, fromPort, toPort *int, cidr string) error { + var one int + err := s.db().QueryRow(`SELECT 1 FROM security_groups WHERE group_id=? AND account_id=?`, groupID, accountID).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return ErrSecurityGroupNotFound + } + if err != nil { + return err + } egressInt := 0 if egress { egressInt = 1 } - _, err := s.db().Exec( + // SQLite IS compares like = but treats NULLs as equal, so nil ports match nil. + res, err := s.db().Exec( `DELETE FROM security_group_rules - WHERE group_id=? AND account_id=? AND egress=? AND protocol=? AND from_port=? AND to_port=? AND cidr=?`, + WHERE group_id=? AND account_id=? AND egress=? AND protocol=? AND from_port IS ? AND to_port IS ? AND cidr=?`, groupID, accountID, egressInt, protocol, fromPort, toPort, cidr) - return err + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrSecurityGroupRuleNotFound + } + return nil } -// SecurityGroupRules returns the rules for a security group. -func (s *EC2Store) SecurityGroupRules(accountID, groupID string) ([]SecurityGroupRule, error) { +// AllSecurityGroupRules returns every rule for the account, bucketed by group +// ID. It lets a DescribeSecurityGroups over N groups run in one query instead +// of N per-group queries. +func (s *EC2Store) AllSecurityGroupRules(accountID string) (map[string][]SecurityGroupRule, error) { rows, err := s.db().Query( `SELECT rule_id, group_id, egress, protocol, from_port, to_port, cidr, account_id - FROM security_group_rules WHERE group_id=? AND account_id=? ORDER BY rule_id`, groupID, accountID) + FROM security_group_rules WHERE account_id=? ORDER BY group_id, rule_id`, accountID) if err != nil { return nil, err } defer func() { _ = rows.Close() }() - var out []SecurityGroupRule + out := make(map[string][]SecurityGroupRule) for rows.Next() { - var r SecurityGroupRule - var egressInt int - if err := rows.Scan(&r.RuleID, &r.GroupID, &egressInt, &r.Protocol, &r.FromPort, &r.ToPort, &r.CIDR, &r.AccountID); err != nil { + r, err := scanSecurityGroupRule(rows) + if err != nil { return nil, err } - r.Egress = egressInt != 0 - out = append(out, r) + out[r.GroupID] = append(out[r.GroupID], r) } return out, rows.Err() } +func scanSecurityGroupRule(rows *sql.Rows) (SecurityGroupRule, error) { + var r SecurityGroupRule + var egressInt int + var from, to sql.NullInt64 + if err := rows.Scan(&r.RuleID, &r.GroupID, &egressInt, &r.Protocol, &from, &to, &r.CIDR, &r.AccountID); err != nil { + return r, err + } + r.Egress = egressInt != 0 + if from.Valid { + v := int(from.Int64) + r.FromPort = &v + } + if to.Valid { + v := int(to.Int64) + r.ToPort = &v + } + return r, nil +} + // ---- Volume operations ---- type Volume struct { diff --git a/tests/compatibility/test_ec2.py b/tests/compatibility/test_ec2.py index fb545b53..86da4ff8 100644 --- a/tests/compatibility/test_ec2.py +++ b/tests/compatibility/test_ec2.py @@ -305,3 +305,177 @@ def test_delete_tags(ec2_client): desc = ec2_client.describe_instances(InstanceIds=[iid]) tags = desc["Reservations"][0]["Instances"][0].get("Tags", []) assert "team" not in {t["Key"] for t in tags} + + +def test_delete_security_group_by_name_removes_one(ec2_client): + # group_name is not unique; deleting by name must remove exactly one group. + name = "dupdel-sg" + ec2_client.create_security_group(GroupName=name, Description="a") + ec2_client.create_security_group(GroupName=name, Description="b") + before = [ + g + for g in ec2_client.describe_security_groups()["SecurityGroups"] + if g["GroupName"] == name + ] + assert len(before) == 2 + ec2_client.delete_security_group(GroupName=name) + after = [ + g + for g in ec2_client.describe_security_groups()["SecurityGroups"] + if g["GroupName"] == name + ] + assert len(after) == 1 + + +def test_describe_security_groups_missing_id_raises(ec2_client): + with pytest.raises(ClientError) as exc: + ec2_client.describe_security_groups(GroupIds=["sg-00000000"]) + assert exc.value.response["Error"]["Code"] == "InvalidGroup.NotFound" + + +def test_describe_addresses_missing_allocation_raises(ec2_client): + with pytest.raises(ClientError) as exc: + ec2_client.describe_addresses(AllocationIds=["eipalloc-00000000"]) + assert exc.value.response["Error"]["Code"] == "InvalidAllocationID.NotFound" + + +def test_revoke_missing_rule_raises(ec2_client): + gid = ec2_client.create_security_group(GroupName="revoke-missing", Description="d")[ + "GroupId" + ] + with pytest.raises(ClientError) as exc: + ec2_client.revoke_security_group_ingress( + GroupId=gid, + IpPermissions=[ + { + "IpProtocol": "tcp", + "FromPort": 22, + "ToPort": 22, + "IpRanges": [{"CidrIp": "0.0.0.0/0"}], + } + ], + ) + assert exc.value.response["Error"]["Code"] == "InvalidPermission.NotFound" + + +def test_revoke_missing_group_raises(ec2_client): + with pytest.raises(ClientError) as exc: + ec2_client.revoke_security_group_ingress( + GroupId="sg-00000000", + IpPermissions=[{"IpProtocol": "-1"}], + ) + assert exc.value.response["Error"]["Code"] == "InvalidGroup.NotFound" + + +def test_batch_stop_is_atomic(ec2_client): + # A bad id later in the batch must not transition the earlier good instance. + good = ec2_client.run_instances(ImageId="ami-atomic", MinCount=1, MaxCount=1)[ + "Instances" + ][0]["InstanceId"] + with pytest.raises(ClientError): + ec2_client.stop_instances(InstanceIds=[good, "i-000000000000bad0"]) + desc = ec2_client.describe_instances(InstanceIds=[good]) + state = desc["Reservations"][0]["Instances"][0]["State"]["Name"] + assert state == "running" + + +def test_associate_address_missing_instance_raises(ec2_client): + alloc_id = ec2_client.allocate_address(Domain="vpc")["AllocationId"] + with pytest.raises(ClientError) as exc: + ec2_client.associate_address( + AllocationId=alloc_id, InstanceId="i-000000000000bad1" + ) + assert exc.value.response["Error"]["Code"] == "InvalidInstanceID.NotFound" + + +def test_authorize_duplicate_rule_raises(ec2_client): + gid = ec2_client.create_security_group(GroupName="dup-rule", Description="d")[ + "GroupId" + ] + perm = [ + { + "IpProtocol": "tcp", + "FromPort": 80, + "ToPort": 80, + "IpRanges": [{"CidrIp": "0.0.0.0/0"}], + } + ] + ec2_client.authorize_security_group_ingress(GroupId=gid, IpPermissions=perm) + with pytest.raises(ClientError) as exc: + ec2_client.authorize_security_group_ingress(GroupId=gid, IpPermissions=perm) + assert exc.value.response["Error"]["Code"] == "InvalidPermission.Duplicate" + + +def test_authorize_and_revoke_by_group_name(ec2_client): + name = "byname-sg" + ec2_client.create_security_group(GroupName=name, Description="d") + perm = [ + { + "IpProtocol": "tcp", + "FromPort": 443, + "ToPort": 443, + "IpRanges": [{"CidrIp": "0.0.0.0/0"}], + } + ] + ec2_client.authorize_security_group_ingress(GroupName=name, IpPermissions=perm) + sg = [ + g + for g in ec2_client.describe_security_groups(GroupNames=[name])[ + "SecurityGroups" + ] + if g["GroupName"] == name + ][0] + assert any(p["FromPort"] == 443 for p in sg["IpPermissions"]) + ec2_client.revoke_security_group_ingress(GroupName=name, IpPermissions=perm) + sg2 = [ + g + for g in ec2_client.describe_security_groups(GroupNames=[name])[ + "SecurityGroups" + ] + if g["GroupName"] == name + ][0] + assert not sg2.get("IpPermissions") + + +def test_all_traffic_rule_omits_ports(ec2_client): + gid = ec2_client.create_security_group(GroupName="alltraffic-sg", Description="d")[ + "GroupId" + ] + ec2_client.authorize_security_group_ingress( + GroupId=gid, + IpPermissions=[{"IpProtocol": "-1", "IpRanges": [{"CidrIp": "0.0.0.0/0"}]}], + ) + sg = ec2_client.describe_security_groups(GroupIds=[gid])["SecurityGroups"][0] + perm = sg["IpPermissions"][0] + assert perm["IpProtocol"] == "-1" + assert "FromPort" not in perm + assert "ToPort" not in perm + + +def test_delete_tags_value_qualified(ec2_client): + iid = ec2_client.run_instances(ImageId="ami-tagval", MinCount=1, MaxCount=1)[ + "Instances" + ][0]["InstanceId"] + ec2_client.create_tags(Resources=[iid], Tags=[{"Key": "team", "Value": "x"}]) + # Value mismatch: nothing removed. + ec2_client.delete_tags(Resources=[iid], Tags=[{"Key": "team", "Value": "y"}]) + tags = ec2_client.describe_instances(InstanceIds=[iid])["Reservations"][0][ + "Instances" + ][0].get("Tags", []) + assert {"Key": "team", "Value": "x"} in tags + # Matching value: removed. + ec2_client.delete_tags(Resources=[iid], Tags=[{"Key": "team", "Value": "x"}]) + tags = ec2_client.describe_instances(InstanceIds=[iid])["Reservations"][0][ + "Instances" + ][0].get("Tags", []) + assert "team" not in {t["Key"] for t in tags} + + +def test_describe_tags_resource_id_filter(ec2_client): + iid = ec2_client.run_instances(ImageId="ami-tagfilter", MinCount=1, MaxCount=1)[ + "Instances" + ][0]["InstanceId"] + ec2_client.create_tags(Resources=[iid], Tags=[{"Key": "owner", "Value": "me"}]) + resp = ec2_client.describe_tags(Filters=[{"Name": "resource-id", "Values": [iid]}]) + assert resp["Tags"] # non-empty + assert all(t["ResourceId"] == iid for t in resp["Tags"]) From 9367303561b372104a70d084eec0b20f0fe12efd Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Mon, 20 Jul 2026 06:01:50 +0900 Subject: [PATCH 4/4] fix(dynamodb): order numeric sort keys by true value via NUMTEXT collation orderByColumn used CAST AS REAL, which loses precision past 2^53 and on high-precision decimals, so numeric ('N') sort/range keys could sort wrong. Register a NUMTEXT SQLite collation (big.Rat comparison, exact across DynamoDB's full 38-digit precision) and order by COLLATE NUMTEXT instead. Also annotates two deliberate mock-appropriate ceilings as accepted (no behaviour change): SQS synchronous redrive and EC2 lenient filter names. Adds a comparator unit test and a collation-wiring test. --- changes/unreleased/Fixed-20260720-055500.yaml | 5 ++ internal/services/dynamodb/store.go | 6 +- internal/services/dynamodb/store_test.go | 35 ++++++++++++ internal/services/sqs/store.go | 5 +- internal/storage/sqlite/store.go | 28 ++++++++- internal/storage/sqlite/store_test.go | 57 +++++++++++++++++++ 6 files changed, 129 insertions(+), 7 deletions(-) create mode 100644 changes/unreleased/Fixed-20260720-055500.yaml diff --git a/changes/unreleased/Fixed-20260720-055500.yaml b/changes/unreleased/Fixed-20260720-055500.yaml new file mode 100644 index 00000000..229e6993 --- /dev/null +++ b/changes/unreleased/Fixed-20260720-055500.yaml @@ -0,0 +1,5 @@ +kind: Fixed +body: 'DynamoDB: numeric (''N'') sort/range keys now order by true numeric value across the full 38-digit precision via a NUMTEXT SQLite collation, instead of a float CAST that lost precision past 2^53' +time: 2026-07-20T05:55:00.000000+09:00 +custom: + Issue: "97" diff --git a/internal/services/dynamodb/store.go b/internal/services/dynamodb/store.go index e1b7c4c4..b13aac8a 100644 --- a/internal/services/dynamodb/store.go +++ b/internal/services/dynamodb/store.go @@ -620,11 +620,11 @@ func indexKeyValue(item Item, idx IndexDef, kind string) (value string, ok bool, // orderByColumn returns the ORDER BY expression for a sort/range key column of // the given attribute type. Numeric ('N') keys must sort numerically, not by the // lexicographic byte order of their TEXT storage (so 2, 10 sorts before 100). -// ponytail: CAST AS REAL loses precision past ~2^53 and on high-precision -// decimals; fine for an emulator, swap for a sortable-number encoding if it bites. +// The NUMTEXT collation compares by true numeric value, exact across DynamoDB's +// full 38-digit precision (a float CAST is lossy past 2^53). func orderByColumn(col, keyType string) string { if keyType == "N" { - return "CAST(" + col + " AS REAL)" + return col + " COLLATE NUMTEXT" } return col } diff --git a/internal/services/dynamodb/store_test.go b/internal/services/dynamodb/store_test.go index a7360aa1..115befb1 100644 --- a/internal/services/dynamodb/store_test.go +++ b/internal/services/dynamodb/store_test.go @@ -119,6 +119,41 @@ func TestDynamoStore_Query(t *testing.T) { assert.Len(t, results, 3) } +// TestDynamoStore_QueryNumericSortKeyPrecision verifies numeric ('N') sort keys +// order by true value even where a float CAST would fail: consecutive integers +// past 2^53 (which collapse to the same IEEE double) and near-equal decimals. +func TestDynamoStore_QueryNumericSortKeyPrecision(t *testing.T) { + store := newTestStore(t) + + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Ledger", + PartitionKey: KeyDef{Name: "Acct", Type: "S"}, + SortKey: &KeyDef{Name: "Seq", Type: "N"}, + })) + + seqs := []string{ + "100", "2", "10", + "9007199254740994", "9007199254740992", "9007199254740993", // 2^53+2, 2^53, 2^53+1 + "1.0000000000000002", "1.0000000000000001", + } + for _, s := range seqs { + require.NoError(t, store.PutItem("Ledger", Item{"Acct": {S: strPtr("a")}, "Seq": {N: strPtr(s)}})) + } + + results, err := store.Query("Ledger", "a", "") + require.NoError(t, err) + + got := make([]string, len(results)) + for i, it := range results { + got[i] = *it["Seq"].N + } + assert.Equal(t, []string{ + "1.0000000000000001", "1.0000000000000002", + "2", "10", "100", + "9007199254740992", "9007199254740993", "9007199254740994", + }, got) +} + // TestDynamoStore_Scan verifies that Scan returns all items in a table. func TestDynamoStore_Scan(t *testing.T) { store := newTestStore(t) diff --git a/internal/services/sqs/store.go b/internal/services/sqs/store.go index 3e07e5cc..a9d4da92 100644 --- a/internal/services/sqs/store.go +++ b/internal/services/sqs/store.go @@ -762,8 +762,9 @@ func (s *QueueStore) StartMessageMoveTask(sourceArn, destArn string, maxRate int } s.mu.Unlock() - // ponytail: redrive runs synchronously with no RUNNING state or rate limiting; - // upgrade to a goroutine + MaxRate throttle if timing fidelity is ever needed. + // ponytail: accepted — redrive runs synchronously with no RUNNING state or rate + // limiting. Deliberate for a mock: synchronous is more test-deterministic than + // async. Revisit only if timing fidelity is ever needed (goroutine + MaxRate throttle). // Lock only one queue at a time (drain the source, then fill the destination) to stay // deadlock-free with the source→DLQ path in ReceiveMessage. now := time.Now() diff --git a/internal/storage/sqlite/store.go b/internal/storage/sqlite/store.go index 04c85a84..e20ee0b8 100644 --- a/internal/storage/sqlite/store.go +++ b/internal/storage/sqlite/store.go @@ -5,13 +5,37 @@ package sqlite import ( "database/sql" "fmt" + "math/big" "os" "path/filepath" "strings" - _ "github.com/mattn/go-sqlite3" + sqlite3 "github.com/mattn/go-sqlite3" ) +// driverName is a sqlite3 driver variant that registers the NUMTEXT collation +// on every connection. NUMTEXT orders TEXT columns holding numeric strings by +// true numeric value (exact across DynamoDB's full 38-digit precision), which +// a float CAST cannot do past 2^53. +const driverName = "sqlite3_numtext" + +func init() { + sql.Register(driverName, &sqlite3.SQLiteDriver{ + ConnectHook: func(conn *sqlite3.SQLiteConn) error { + return conn.RegisterCollation("NUMTEXT", compareNumericText) + }, + }) +} + +func compareNumericText(a, b string) int { + ra, oka := new(big.Rat).SetString(a) + rb, okb := new(big.Rat).SetString(b) + if oka && okb { + return ra.Cmp(rb) + } + return strings.Compare(a, b) // non-numeric values fall back to byte order +} + type Migration struct { Version int SQL string @@ -32,7 +56,7 @@ func Open(dbPath string, migrations []Migration) (*Store, error) { return nil, fmt.Errorf("create data dir: %w", err) } - db, err := sql.Open("sqlite3", dbPath) + db, err := sql.Open(driverName, dbPath) if err != nil { return nil, fmt.Errorf("open sqlite: %w", err) } diff --git a/internal/storage/sqlite/store_test.go b/internal/storage/sqlite/store_test.go index 805c473c..1802d562 100644 --- a/internal/storage/sqlite/store_test.go +++ b/internal/storage/sqlite/store_test.go @@ -58,3 +58,60 @@ func TestOpen_AppliesNewMigrations(t *testing.T) { _, err = store2.DB().Exec(`INSERT INTO items (id, name, description) VALUES ('1', 'test', 'desc')`) assert.NoError(t, err) } + +// TestCompareNumericText guards the NUMTEXT comparator on the cases a float CAST +// gets wrong: consecutive integers past 2^53 (which collapse to one double) and +// near-equal high-precision decimals. It also covers numeric-vs-text ordering, +// signs, and the non-numeric byte-order fallback. +func TestCompareNumericText(t *testing.T) { + cases := []struct { + a, b string + want int + }{ + {"9007199254740992", "9007199254740993", -1}, // 2^53 vs 2^53+1 (float: tie) + {"9007199254740994", "9007199254740993", 1}, // 2^53+2 vs 2^53+1 (float: tie) + {"1.0000000000000001", "1.0000000000000002", -1}, + {"2", "10", -1}, // numeric, not lexicographic + {"10", "2", 1}, + {"-5", "3", -1}, + {"100", "100", 0}, + {"abc", "abd", -1}, // non-numeric → byte order + {"5", "x", -1}, // one side non-numeric → byte order + } + for _, c := range cases { + assert.Equalf(t, c.want, compareNumericText(c.a, c.b), "compareNumericText(%q, %q)", c.a, c.b) + } +} + +// TestNumtextCollationOrdering proves the collation is registered on the +// connection and changes ORDER BY: plain TEXT order sorts "10","100","2" but +// COLLATE NUMTEXT sorts them numerically. +func TestNumtextCollationOrdering(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + store, err := Open(dbPath, []Migration{ + {Version: 1, SQL: `CREATE TABLE nums (v TEXT NOT NULL)`}, + }) + require.NoError(t, err) + defer func() { _ = store.Close() }() + + for _, v := range []string{"100", "2", "10"} { + _, err := store.DB().Exec(`INSERT INTO nums (v) VALUES (?)`, v) + require.NoError(t, err) + } + + read := func(orderBy string) []string { + rows, err := store.DB().Query(`SELECT v FROM nums ORDER BY ` + orderBy) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + var out []string + for rows.Next() { + var v string + require.NoError(t, rows.Scan(&v)) + out = append(out, v) + } + return out + } + + assert.Equal(t, []string{"10", "100", "2"}, read("v")) // byte order + assert.Equal(t, []string{"2", "10", "100"}, read("v COLLATE NUMTEXT")) // numeric order +}