AWS Step Functions is a serverless orchestration service that lets you coordinate multiple AWS services into serverless workflows. This implementation focuses on securing Step Functions workflows in Java applications.
Step Functions Security Overview
Key security aspects:
- IAM roles and policies for execution permissions
- Encryption for data at rest and in transit
- Input/Output validation and sanitization
- Access control and resource-based policies
- Audit logging and monitoring
- VPC endpoints for private network access
Dependencies and Setup
Maven Configuration:
<dependencies> <!-- AWS SDK v2 --> <dependency> <groupId>software.amazon.awssdk</groupId> <artifactId>stepfunctions</artifactId> <version>2.20.0</version> </dependency> <dependency> <groupId>software.amazon.awssdk</groupId> <artifactId>sts</artifactId> <version>2.20.0</version> </dependency> <dependency> <groupId>software.amazon.awssdk</groupId> <artifactId>kms</artifactId> <version>2.20.0</version> </dependency> <dependency> <groupId>software.amazon.awssdk</groupId> <artifactId>secretsmanager</artifactId> <version>2.20.0</version> </dependency> <!-- Encryption --> <dependency> <groupId>org.bouncycastle</groupId> <artifactId>bcprov-jdk18on</artifactId> <version>1.76</version> </dependency> <!-- JSON Processing --> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.15.2</version> </dependency> <!-- Validation --> <dependency> <groupId>jakarta.validation</groupId> <artifactId>jakarta.validation-api</artifactId> <version>3.0.2</version> </dependency> <dependency> <groupId>org.hibernate.validator</groupId> <artifactId>hibernate-validator</artifactId> <version>8.0.0.Final</version> </dependency> <!-- Spring Boot --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-security</artifactId> </dependency> <!-- Logging --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>2.0.7</version> </dependency> </dependencies>
Configuration Management
StepFunctionsConfig.java:
@Configuration
@ConfigurationProperties(prefix = "aws.step-functions")
@Data
public class StepFunctionsConfig {
private String region = "us-east-1";
private String endpointOverride;
private ExecutionConfig execution = new ExecutionConfig();
private SecurityConfig security = new SecurityConfig();
private MonitoringConfig monitoring = new MonitoringConfig();
@Data
public static class ExecutionConfig {
private String defaultStateMachineArn;
private int maxExecutionHistorySize = 1000;
private Duration executionTimeout = Duration.ofHours(24);
private boolean enableXRayTracing = true;
private boolean enableCloudWatchLogs = true;
}
@Data
public static class SecurityConfig {
private String kmsKeyId;
private String encryptionKeyAlias;
private boolean enforceInputValidation = true;
private boolean enableVpcEndpoints = false;
private List<String> allowedIamRoles = new ArrayList<>();
private List<String> restrictedStates = new ArrayList<>();
private int maxInputSizeBytes = 1024 * 1024; // 1MB
}
@Data
public static class MonitoringConfig {
private boolean enableCloudWatchMetrics = true;
private boolean enableDetailedMetrics = true;
private String logGroupName = "/aws/stepfunctions";
private Duration auditLogRetention = Duration.ofDays(365);
}
public StepFunctionsClient createStepFunctionsClient() {
StepFunctionsClientBuilder builder = StepFunctionsClient.builder()
.region(Region.of(region));
if (endpointOverride != null) {
builder.endpointOverride(URI.create(endpointOverride));
}
return builder.build();
}
}
IAM Policy Management
IamPolicyManager.java:
@Service
@Slf4j
public class IamPolicyManager {
private final ObjectMapper objectMapper;
public IamPolicyManager() {
this.objectMapper = new ObjectMapper();
}
public String createLeastPrivilegePolicy(String stateMachineArn,
List<String> allowedActions,
Map<String, String> resourceConstraints) {
try {
JsonNode policy = createPolicyDocument(stateMachineArn, allowedActions, resourceConstraints);
return objectMapper.writeValueAsString(policy);
} catch (Exception e) {
throw new PolicyException("Failed to create IAM policy", e);
}
}
public String createScopedExecutionPolicy(String stateMachineArn,
String executionId,
Map<String, Object> context) {
try {
JsonNode policy = objectMapper.createObjectNode()
.set("Version", objectMapper.valueToTree("2012-10-17"))
.set("Statement", createScopedStatements(stateMachineArn, executionId, context));
return objectMapper.writeValueAsString(policy);
} catch (Exception e) {
throw new PolicyException("Failed to create scoped execution policy", e);
}
}
public boolean validatePolicy(String policyJson) {
try {
JsonNode policy = objectMapper.readTree(policyJson);
// Basic policy validation
if (!policy.has("Version") || !policy.has("Statement")) {
return false;
}
JsonNode statements = policy.get("Statement");
if (!statements.isArray()) {
return false;
}
// Check for dangerous permissions
for (JsonNode statement : statements) {
if (hasDangerousPermissions(statement)) {
return false;
}
}
return true;
} catch (Exception e) {
log.warn("Policy validation failed", e);
return false;
}
}
private JsonNode createPolicyDocument(String stateMachineArn,
List<String> allowedActions,
Map<String, String> resourceConstraints) {
return objectMapper.createObjectNode()
.set("Version", objectMapper.valueToTree("2012-10-17"))
.set("Statement", createPolicyStatements(stateMachineArn, allowedActions, resourceConstraints));
}
private JsonNode createPolicyStatements(String stateMachineArn,
List<String> allowedActions,
Map<String, String> resourceConstraints) {
ArrayNode statements = objectMapper.createArrayNode();
// Allow specific Step Functions actions
ObjectNode stepFunctionsStatement = objectMapper.createObjectNode();
stepFunctionsStatement.put("Effect", "Allow");
stepFunctionsStatement.set("Action", objectMapper.valueToTree(allowedActions));
stepFunctionsStatement.set("Resource", objectMapper.valueToTree(List.of(stateMachineArn)));
// Add resource constraints if provided
if (resourceConstraints != null && !resourceConstraints.isEmpty()) {
stepFunctionsStatement.set("Condition", createConditions(resourceConstraints));
}
statements.add(stepFunctionsStatement);
// Add necessary service actions
statements.add(createServiceStatement("logs", List.of(
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
)));
statements.add(createServiceStatement("xray", List.of(
"xray:PutTraceSegments",
"xray:PutTelemetryRecords"
)));
return statements;
}
private JsonNode createScopedStatements(String stateMachineArn,
String executionId,
Map<String, Object> context) {
ArrayNode statements = objectMapper.createArrayNode();
// Allow execution-specific actions
ObjectNode executionStatement = objectMapper.createObjectNode();
executionStatement.put("Effect", "Allow");
executionStatement.set("Action", objectMapper.valueToTree(List.of(
"states:DescribeExecution",
"states:GetExecutionHistory"
)));
executionStatement.set("Resource", objectMapper.valueToTree(
stateMachineArn + ":execution:" + executionId));
// Add context-based conditions
if (context != null && !context.isEmpty()) {
executionStatement.set("Condition", createContextConditions(context));
}
statements.add(executionStatement);
return statements;
}
private JsonNode createServiceStatement(String service, List<String> actions) {
ObjectNode statement = objectMapper.createObjectNode();
statement.put("Effect", "Allow");
statement.set("Action", objectMapper.valueToTree(actions));
statement.set("Resource", objectMapper.valueToTree("*"));
return statement;
}
private JsonNode createConditions(Map<String, String> constraints) {
ObjectNode conditions = objectMapper.createObjectNode();
constraints.forEach((key, value) -> {
ObjectNode condition = objectMapper.createObjectNode();
condition.set(key, objectMapper.valueToTree(List.of(value)));
conditions.set("StringEquals", condition);
});
return conditions;
}
private JsonNode createContextConditions(Map<String, Object> context) {
ObjectNode conditions = objectMapper.createObjectNode();
context.forEach((key, value) -> {
if (value instanceof String) {
ObjectNode condition = objectMapper.createObjectNode();
condition.set(key, objectMapper.valueToTree(List.of(value)));
conditions.set("StringEquals", condition);
}
});
return conditions;
}
private boolean hasDangerousPermissions(JsonNode statement) {
if (!"Allow".equals(statement.get("Effect").asText())) {
return false;
}
JsonNode actions = statement.get("Action");
if (actions != null) {
for (JsonNode action : actions) {
String actionStr = action.asText();
if (actionStr.equals("*") ||
actionStr.startsWith("iam:") ||
actionStr.startsWith("sts:") ||
actionStr.contains("PassRole") ||
actionStr.contains("PutRolePolicy")) {
return true;
}
}
}
JsonNode resources = statement.get("Resource");
if (resources != null) {
for (JsonNode resource : resources) {
if ("*".equals(resource.asText())) {
return true;
}
}
}
return false;
}
}
Input Validation and Sanitization
InputValidator.java:
@Service
@Slf4j
public class InputValidator {
private final ObjectMapper objectMapper;
private final Validator validator;
private final Set<String> allowedJsonPaths;
public InputValidator() {
this.objectMapper = new ObjectMapper();
this.validator = Validation.buildDefaultValidatorFactory().getValidator();
this.allowedJsonPaths = initializeAllowedJsonPaths();
}
public ValidationResult validateExecutionInput(String input,
Class<?> inputType,
ValidationContext context) {
try {
List<ValidationError> errors = new ArrayList<>();
// Check input size
if (input.getBytes().length > context.getMaxInputSize()) {
errors.add(new ValidationError("INPUT_SIZE",
"Input exceeds maximum allowed size: " + context.getMaxInputSize()));
}
// Parse and validate JSON
JsonNode jsonInput = objectMapper.readTree(input);
// Validate against schema
errors.addAll(validateJsonSchema(jsonInput, context.getSchema()));
// Validate sensitive data
errors.addAll(validateSensitiveData(jsonInput));
// Validate against input type constraints
if (inputType != null) {
errors.addAll(validateInputType(jsonInput, inputType));
}
// Sanitize input if validation passes
JsonNode sanitizedInput = errors.isEmpty() ?
sanitizeInput(jsonInput, context) : jsonInput;
return ValidationResult.builder()
.valid(errors.isEmpty())
.errors(errors)
.originalInput(input)
.sanitizedInput(objectMapper.writeValueAsString(sanitizedInput))
.build();
} catch (Exception e) {
log.error("Input validation failed", e);
return ValidationResult.builder()
.valid(false)
.errors(List.of(new ValidationError("VALIDATION_ERROR",
"Input validation failed: " + e.getMessage())))
.originalInput(input)
.build();
}
}
public ValidationResult validateStateInput(String input,
String stateName,
Map<String, Object> stateConstraints) {
try {
List<ValidationError> errors = new ArrayList<>();
JsonNode jsonInput = objectMapper.readTree(input);
// State-specific validation
if (stateConstraints != null) {
errors.addAll(validateStateConstraints(jsonInput, stateConstraints));
}
// Check for prohibited patterns
errors.addAll(validateProhibitedPatterns(jsonInput));
return ValidationResult.builder()
.valid(errors.isEmpty())
.errors(errors)
.originalInput(input)
.sanitizedInput(objectMapper.writeValueAsString(sanitizeStateInput(jsonInput, stateName)))
.build();
} catch (Exception e) {
throw new ValidationException("State input validation failed", e);
}
}
private List<ValidationError> validateJsonSchema(JsonNode input, JsonNode schema) {
List<ValidationError> errors = new ArrayList<>();
if (schema == null) {
return errors; // No schema defined
}
try {
// Use JSON Schema validator (implementation would use library like networknt/json-schema-validator)
// This is a simplified version
if (!input.isObject() && schema.get("type").asText().equals("object")) {
errors.add(new ValidationError("SCHEMA_TYPE", "Input must be a JSON object"));
}
// Add more schema validation logic...
} catch (Exception e) {
errors.add(new ValidationError("SCHEMA_VALIDATION",
"Schema validation error: " + e.getMessage()));
}
return errors;
}
private List<ValidationError> validateSensitiveData(JsonNode input) {
List<ValidationError> errors = new ArrayList<>();
// Check for sensitive data patterns
detectSensitiveData(input, "", errors);
return errors;
}
private void detectSensitiveData(JsonNode node, String path, List<ValidationError> errors) {
if (node.isObject()) {
node.fields().forEachRemaining(entry -> {
String fieldName = entry.getKey();
JsonNode fieldValue = entry.getValue();
String currentPath = path.isEmpty() ? fieldName : path + "." + fieldName;
// Check field name for sensitive indicators
if (isSensitiveField(fieldName)) {
errors.add(new ValidationError("SENSITIVE_FIELD",
"Potential sensitive field detected: " + currentPath));
}
// Check field value for sensitive patterns
if (fieldValue.isTextual()) {
String value = fieldValue.asText();
if (containsSensitivePattern(value)) {
errors.add(new ValidationError("SENSITIVE_DATA",
"Potential sensitive data detected in: " + currentPath));
}
}
// Recursively check nested objects
detectSensitiveData(fieldValue, currentPath, errors);
});
} else if (node.isArray()) {
for (int i = 0; i < node.size(); i++) {
detectSensitiveData(node.get(i), path + "[" + i + "]", errors);
}
}
}
private List<ValidationError> validateInputType(JsonNode input, Class<?> inputType) {
List<ValidationError> errors = new ArrayList<>();
try {
Object typedInput = objectMapper.treeToValue(input, inputType);
Set<ConstraintViolation<Object>> violations = validator.validate(typedInput);
violations.forEach(violation ->
errors.add(new ValidationError(
"TYPE_VALIDATION",
violation.getPropertyPath() + ": " + violation.getMessage())));
} catch (Exception e) {
errors.add(new ValidationError("TYPE_CONVERSION",
"Failed to convert input to type: " + e.getMessage()));
}
return errors;
}
private List<ValidationError> validateStateConstraints(JsonNode input,
Map<String, Object> constraints) {
List<ValidationError> errors = new ArrayList<>();
constraints.forEach((key, value) -> {
JsonNode node = input.at(key);
if (!node.isMissingNode()) {
// Validate constraint
if (value instanceof Number) {
if (node.isNumber() && node.decimalValue().compareTo(
BigDecimal.valueOf(((Number) value).doubleValue())) > 0) {
errors.add(new ValidationError("CONSTRAINT_VIOLATION",
key + " exceeds maximum value: " + value));
}
} else if (value instanceof String) {
String pattern = (String) value;
if (node.isTextual() && !node.asText().matches(pattern)) {
errors.add(new ValidationError("CONSTRAINT_VIOLATION",
key + " does not match pattern: " + pattern));
}
}
}
});
return errors;
}
private List<ValidationError> validateProhibitedPatterns(JsonNode input) {
List<ValidationError> errors = new ArrayList<>();
// Check for SQL injection patterns
checkForPatterns(input, "",
Pattern.compile("(?i)(SELECT|INSERT|UPDATE|DELETE|DROP|UNION|OR|AND)"),
"SQL_INJECTION", errors);
// Check for command injection patterns
checkForPatterns(input, "",
Pattern.compile("[|&;`$<>]"),
"COMMAND_INJECTION", errors);
// Check for path traversal patterns
checkForPatterns(input, "",
Pattern.compile("\\.\\./|\\.\\.\\\\"),
"PATH_TRAVERSAL", errors);
return errors;
}
private void checkForPatterns(JsonNode node, String path, Pattern pattern,
String errorCode, List<ValidationError> errors) {
if (node.isTextual()) {
String value = node.asText();
if (pattern.matcher(value).find()) {
errors.add(new ValidationError(errorCode,
"Prohibited pattern detected in: " + (path.isEmpty() ? "root" : path)));
}
} else if (node.isObject()) {
node.fields().forEachRemaining(entry ->
checkForPatterns(entry.getValue(),
path.isEmpty() ? entry.getKey() : path + "." + entry.getKey(),
pattern, errorCode, errors));
} else if (node.isArray()) {
for (int i = 0; i < node.size(); i++) {
checkForPatterns(node.get(i), path + "[" + i + "]", pattern, errorCode, errors);
}
}
}
private JsonNode sanitizeInput(JsonNode input, ValidationContext context) {
try {
// Remove sensitive fields
JsonNode sanitized = removeSensitiveFields(input, context.getSensitiveFields());
// Truncate large fields
sanitized = truncateLargeFields(sanitized, context.getMaxFieldSizes());
// Escape special characters
sanitized = escapeSpecialCharacters(sanitized);
return sanitized;
} catch (Exception e) {
log.warn("Input sanitization failed, returning original input", e);
return input;
}
}
private JsonNode sanitizeStateInput(JsonNode input, String stateName) {
// State-specific sanitization
JsonNode sanitized = input.deepCopy();
// Remove fields not allowed for this state
if (allowedJsonPaths.contains(stateName)) {
sanitized = filterAllowedPaths(sanitized, stateName);
}
return sanitized;
}
private JsonNode removeSensitiveFields(JsonNode node, Set<String> sensitiveFields) {
if (node.isObject()) {
ObjectNode objectNode = (ObjectNode) node;
sensitiveFields.forEach(objectNode::remove);
return objectNode;
}
return node;
}
private JsonNode truncateLargeFields(JsonNode node, Map<String, Integer> maxSizes) {
if (node.isObject()) {
ObjectNode objectNode = (ObjectNode) node;
maxSizes.forEach((field, maxSize) -> {
if (objectNode.has(field) && objectNode.get(field).isTextual()) {
String value = objectNode.get(field).asText();
if (value.length() > maxSize) {
objectNode.put(field, value.substring(0, maxSize) + "...");
}
}
});
return objectNode;
}
return node;
}
private JsonNode escapeSpecialCharacters(JsonNode node) {
// Implementation would escape HTML, JSON, etc.
return node;
}
private JsonNode filterAllowedPaths(JsonNode node, String stateName) {
// Implementation would filter JSON based on allowed paths for the state
return node;
}
private boolean isSensitiveField(String fieldName) {
String lowerField = fieldName.toLowerCase();
return lowerField.contains("password") ||
lowerField.contains("secret") ||
lowerField.contains("key") ||
lowerField.contains("token") ||
lowerField.contains("credential") ||
lowerField.contains("auth");
}
private boolean containsSensitivePattern(String value) {
// Check for common sensitive data patterns
return value.matches(".*[A-Za-z0-9+/]{40,}.*") || // Base64-like
value.matches(".*[0-9]{9,}.*") || // Potential SSN/credit card
value.matches(".*[a-fA-F0-9]{32,}.*"); // MD5 hash-like
}
private Set<String> initializeAllowedJsonPaths() {
return Set.of(
"PaymentState./amount",
"PaymentState./currency",
"UserState./userId",
"UserState./email"
);
}
@Data
@Builder
public static class ValidationResult {
private boolean valid;
private List<ValidationError> errors;
private String originalInput;
private String sanitizedInput;
}
@Data
@AllArgsConstructor
public static class ValidationError {
private String code;
private String message;
}
@Data
@Builder
public static class ValidationContext {
private int maxInputSize;
private JsonNode schema;
private Set<String> sensitiveFields;
private Map<String, Integer> maxFieldSizes;
}
}
Secure Step Functions Client
SecureStepFunctionsClient.java:
@Service
@Slf4j
public class SecureStepFunctionsClient {
private final StepFunctionsConfig config;
private final InputValidator inputValidator;
private final EncryptionService encryptionService;
private final AuditLogger auditLogger;
private final StepFunctionsClient client;
private final KmsClient kmsClient;
public SecureStepFunctionsClient(StepFunctionsConfig config,
InputValidator inputValidator,
EncryptionService encryptionService,
AuditLogger auditLogger) {
this.config = config;
this.inputValidator = inputValidator;
this.encryptionService = encryptionService;
this.auditLogger = auditLogger;
this.client = config.createStepFunctionsClient();
this.kmsClient = KmsClient.builder()
.region(Region.of(config.getRegion()))
.build();
}
public SecureExecution startSecureExecution(StartExecutionRequest request,
SecurityContext securityContext) {
try {
// Validate input
InputValidator.ValidationResult validation = inputValidator.validateExecutionInput(
request.input(), null, createValidationContext());
if (!validation.isValid()) {
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("INPUT_VALIDATION_FAILED")
.resourceArn(request.stateMachineArn())
.principal(securityContext.getPrincipal())
.details(Map.of("errors", validation.getErrors()))
.build());
throw new ValidationException("Input validation failed: " + validation.getErrors());
}
// Encrypt sensitive data in input
String encryptedInput = encryptionService.encryptSensitiveData(
validation.getSanitizedInput(), securityContext);
// Create secure execution request
StartExecutionRequest secureRequest = request.toBuilder()
.input(encryptedInput)
.build();
// Start execution
StartExecutionResponse response = client.startExecution(secureRequest);
// Log audit event
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("EXECUTION_STARTED")
.resourceArn(request.stateMachineArn())
.executionArn(response.executionArn())
.principal(securityContext.getPrincipal())
.details(Map.of(
"inputSize", encryptedInput.length(),
"validated", true
))
.build());
return SecureExecution.builder()
.executionArn(response.executionArn())
.startDate(response.startDate())
.securityContext(securityContext)
.validationResult(validation)
.build();
} catch (Exception e) {
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("EXECUTION_FAILED")
.resourceArn(request.stateMachineArn())
.principal(securityContext.getPrincipal())
.details(Map.of("error", e.getMessage()))
.severity(Severity.ERROR)
.build());
throw new ExecutionException("Secure execution failed", e);
}
}
public DescribeExecutionResponse describeSecureExecution(String executionArn,
SecurityContext securityContext) {
try {
// Verify access permissions
if (!hasExecutionAccess(executionArn, securityContext)) {
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("UNAUTHORIZED_ACCESS")
.executionArn(executionArn)
.principal(securityContext.getPrincipal())
.severity(Severity.HIGH)
.build());
throw new AccessDeniedException("Access denied to execution: " + executionArn);
}
DescribeExecutionResponse response = client.describeExecution(
DescribeExecutionRequest.builder()
.executionArn(executionArn)
.build());
// Decrypt output if present
String decryptedOutput = response.output() != null ?
encryptionService.decryptData(response.output(), securityContext) : null;
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("EXECUTION_ACCESSED")
.executionArn(executionArn)
.principal(securityContext.getPrincipal())
.build());
return response.toBuilder()
.output(decryptedOutput)
.build();
} catch (Exception e) {
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("EXECUTION_ACCESS_FAILED")
.executionArn(executionArn)
.principal(securityContext.getPrincipal())
.details(Map.of("error", e.getMessage()))
.build());
throw new ExecutionException("Failed to describe execution", e);
}
}
public GetExecutionHistoryResponse getSecureExecutionHistory(String executionArn,
SecurityContext securityContext,
HistoryFilter filter) {
try {
// Verify access permissions
if (!hasExecutionAccess(executionArn, securityContext)) {
throw new AccessDeniedException("Access denied to execution history: " + executionArn);
}
GetExecutionHistoryRequest request = GetExecutionHistoryRequest.builder()
.executionArn(executionArn)
.maxResults(filter.getMaxResults())
.reverseOrder(filter.isReverseOrder())
.build();
GetExecutionHistoryResponse response = client.getExecutionHistory(request);
// Sanitize history events
List<HistoryEvent> sanitizedEvents = response.events().stream()
.map(this::sanitizeHistoryEvent)
.collect(Collectors.toList());
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("HISTORY_ACCESSED")
.executionArn(executionArn)
.principal(securityContext.getPrincipal())
.details(Map.of("eventCount", sanitizedEvents.size()))
.build());
return response.toBuilder()
.events(sanitizedEvents)
.build();
} catch (Exception e) {
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("HISTORY_ACCESS_FAILED")
.executionArn(executionArn)
.principal(securityContext.getPrincipal())
.details(Map.of("error", e.getMessage()))
.build());
throw new ExecutionException("Failed to get execution history", e);
}
}
public void stopSecureExecution(String executionArn,
String cause,
SecurityContext securityContext) {
try {
// Verify stop permissions
if (!hasStopPermission(executionArn, securityContext)) {
throw new AccessDeniedException("Stop permission denied for execution: " + executionArn);
}
client.stopExecution(StopExecutionRequest.builder()
.executionArn(executionArn)
.cause(cause)
.build());
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("EXECUTION_STOPPED")
.executionArn(executionArn)
.principal(securityContext.getPrincipal())
.details(Map.of("cause", cause))
.build());
} catch (Exception e) {
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("EXECUTION_STOP_FAILED")
.executionArn(executionArn)
.principal(securityContext.getPrincipal())
.details(Map.of("error", e.getMessage()))
.build());
throw new ExecutionException("Failed to stop execution", e);
}
}
public List<SecureExecution> listSecureExecutions(String stateMachineArn,
SecurityContext securityContext,
ExecutionFilter filter) {
try {
// Apply security filters
ListExecutionsRequest request = applySecurityFilters(
ListExecutionsRequest.builder()
.stateMachineArn(stateMachineArn)
.maxResults(filter.getMaxResults())
.statusFilter(filter.getStatusFilter())
.build(),
securityContext);
ListExecutionsResponse response = client.listExecutions(request);
// Filter results based on security context
List<ExecutionListItem> secureExecutions = response.executions().stream()
.filter(execution -> hasExecutionAccess(execution.executionArn(), securityContext))
.collect(Collectors.toList());
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("EXECUTIONS_LISTED")
.resourceArn(stateMachineArn)
.principal(securityContext.getPrincipal())
.details(Map.of("count", secureExecutions.size()))
.build());
return secureExecutions.stream()
.map(execution -> SecureExecution.builder()
.executionArn(execution.executionArn())
.stateMachineArn(execution.stateMachineArn())
.name(execution.name())
.status(execution.status())
.startDate(execution.startDate())
.stopDate(execution.stopDate())
.securityContext(securityContext)
.build())
.collect(Collectors.toList());
} catch (Exception e) {
auditLogger.logSecurityEvent(SecurityEvent.builder()
.eventType("EXECUTIONS_LIST_FAILED")
.resourceArn(stateMachineArn)
.principal(securityContext.getPrincipal())
.details(Map.of("error", e.getMessage()))
.build());
throw new ExecutionException("Failed to list executions", e);
}
}
private boolean hasExecutionAccess(String executionArn, SecurityContext securityContext) {
// Implement access control logic
if (securityContext.isAdmin()) {
return true;
}
// Check if principal owns the execution
if (executionArn.contains(securityContext.getPrincipal())) {
return true;
}
// Check resource-based policies
return checkResourcePolicy(executionArn, securityContext);
}
private boolean hasStopPermission(String executionArn, SecurityContext securityContext) {
// Only admins and execution owners can stop executions
return securityContext.isAdmin() || executionArn.contains(securityContext.getPrincipal());
}
private boolean checkResourcePolicy(String resourceArn, SecurityContext securityContext) {
// Implement resource-based policy checking
// This would typically check IAM policies and resource tags
return true; // Simplified for example
}
private ListExecutionsRequest applySecurityFilters(ListExecutionsRequest request,
SecurityContext securityContext) {
// Apply security-based filters to the request
if (!securityContext.isAdmin()) {
// Non-admins can only see their own executions
// This would typically use resource tags or naming conventions
}
return request;
}
private HistoryEvent sanitizeHistoryEvent(HistoryEvent event) {
// Remove sensitive data from history events
return event.toBuilder()
.build(); // Implementation would sanitize event details
}
private InputValidator.ValidationContext createValidationContext() {
return InputValidator.ValidationContext.builder()
.maxInputSize(config.getSecurity().getMaxInputSizeBytes())
.sensitiveFields(Set.of("password", "secret", "token"))
.maxFieldSizes(Map.of("description", 500, "notes", 1000))
.build();
}
}
Encryption Service
EncryptionService.java:
@Service
@Slf4j
public class EncryptionService {
private final StepFunctionsConfig config;
private final KmsClient kmsClient;
private final ObjectMapper objectMapper;
public EncryptionService(StepFunctionsConfig config) {
this.config = config;
this.kmsClient = KmsClient.builder()
.region(Region.of(config.getRegion()))
.build();
this.objectMapper = new ObjectMapper();
}
public String encryptSensitiveData(String input, SecurityContext securityContext) {
try {
JsonNode jsonInput = objectMapper.readTree(input);
JsonNode encrypted = encryptJsonNode(jsonInput, securityContext);
return objectMapper.writeValueAsString(encrypted);
} catch (Exception e) {
throw new EncryptionException("Failed to encrypt sensitive data", e);
}
}
public String decryptData(String encryptedData, SecurityContext securityContext) {
try {
JsonNode jsonData = objectMapper.readTree(encryptedData);
JsonNode decrypted = decryptJsonNode(jsonData, securityContext);
return objectMapper.writeValueAsString(decrypted);
} catch (Exception e) {
throw new EncryptionException("Failed to decrypt data", e);
}
}
public String encryptField(String plaintext, String fieldName, SecurityContext context) {
try {
EncryptRequest request = EncryptRequest.builder()
.keyId(config.getSecurity().getKmsKeyId())
.plaintext(SdkBytes.fromUtf8String(plaintext))
.encryptionContext(createEncryptionContext(fieldName, context))
.build();
EncryptResponse response = kmsClient.encrypt(request);
return Base64.getEncoder().encodeToString(response.ciphertextBlob().asByteArray());
} catch (Exception e) {
throw new EncryptionException("Failed to encrypt field: " + fieldName, e);
}
}
public String decryptField(String encryptedBase64, String fieldName, SecurityContext context) {
try {
byte[] ciphertext = Base64.getDecoder().decode(encryptedBase64);
DecryptRequest request = DecryptRequest.builder()
.ciphertextBlob(SdkBytes.fromByteArray(ciphertext))
.encryptionContext(createEncryptionContext(fieldName, context))
.build();
DecryptResponse response = kmsClient.decrypt(request);
return response.plaintext().asUtf8String();
} catch (Exception e) {
throw new EncryptionException("Failed to decrypt field: " + fieldName, e);
}
}
private JsonNode encryptJsonNode(JsonNode node, SecurityContext context) {
if (node.isObject()) {
ObjectNode objectNode = (ObjectNode) node;
Iterator<Map.Entry<String, JsonNode>> fields = objectNode.fields();
while (fields.hasNext()) {
Map.Entry<String, JsonNode> field = fields.next();
String fieldName = field.getKey();
JsonNode fieldValue = field.getValue();
if (isSensitiveField(fieldName)) {
if (fieldValue.isTextual()) {
String encrypted = encryptField(fieldValue.asText(), fieldName, context);
objectNode.put(fieldName, "encrypted:" + encrypted);
}
} else if (fieldValue.isObject() || fieldValue.isArray()) {
encryptJsonNode(fieldValue, context);
}
}
} else if (node.isArray()) {
ArrayNode arrayNode = (ArrayNode) node;
for (int i = 0; i < arrayNode.size(); i++) {
JsonNode element = arrayNode.get(i);
if (element.isObject() || element.isArray()) {
encryptJsonNode(element, context);
}
}
}
return node;
}
private JsonNode decryptJsonNode(JsonNode node, SecurityContext context) {
if (node.isObject()) {
ObjectNode objectNode = (ObjectNode) node;
Iterator<Map.Entry<String, JsonNode>> fields = objectNode.fields();
while (fields.hasNext()) {
Map.Entry<String, JsonNode> field = fields.next();
String fieldName = field.getKey();
JsonNode fieldValue = field.getValue();
if (fieldValue.isTextual()) {
String value = fieldValue.asText();
if (value.startsWith("encrypted:")) {
String encrypted = value.substring("encrypted:".length());
String decrypted = decryptField(encrypted, fieldName, context);
objectNode.put(fieldName, decrypted);
}
} else if (fieldValue.isObject() || fieldValue.isArray()) {
decryptJsonNode(fieldValue, context);
}
}
} else if (node.isArray()) {
ArrayNode arrayNode = (ArrayNode) node;
for (int i = 0; i < arrayNode.size(); i++) {
JsonNode element = arrayNode.get(i);
if (element.isObject() || element.isArray()) {
decryptJsonNode(element, context);
}
}
}
return node;
}
private Map<String, String> createEncryptionContext(String fieldName, SecurityContext context) {
Map<String, String> encryptionContext = new HashMap<>();
encryptionContext.put("field", fieldName);
encryptionContext.put("principal", context.getPrincipal());
encryptionContext.put("timestamp", Instant.now().toString());
encryptionContext.put("purpose", "step-functions-security");
return encryptionContext;
}
private boolean isSensitiveField(String fieldName) {
String lowerField = fieldName.toLowerCase();
return lowerField.contains("password") ||
lowerField.contains("secret") ||
lowerField.contains("key") ||
lowerField.contains("token") ||
lowerField.contains("credential");
}
}
Audit Logging
AuditLogger.java:
@Service
@Slf4j
public class AuditLogger {
private final CloudWatchLogsClient cloudWatchLogsClient;
private final String logGroupName;
private final String logStreamName;
public AuditLogger(StepFunctionsConfig config) {
this.cloudWatchLogsClient = CloudWatchLogsClient.builder()
.region(Region.of(config.getRegion()))
.build();
this.logGroupName = config.getMonitoring().getLogGroupName() + "-audit";
this.logStreamName = "step-functions-security-" + Instant.now().getEpochSecond();
initializeLogStream();
}
public void logSecurityEvent(SecurityEvent event) {
try {
String logMessage = createLogMessage(event);
PutLogEventsRequest request = PutLogEventsRequest.builder()
.logGroupName(logGroupName)
.logStreamName(logStreamName)
.logEvents(LogEvent.builder()
.message(logMessage)
.timestamp(System.currentTimeMillis())
.build())
.build();
cloudWatchLogsClient.putLogEvents(request);
// Also log to application logs
log.info("Security Event: {}", logMessage);
} catch (Exception e) {
log.error("Failed to log security event", e);
// Fallback to application logs
log.warn("Security Event (Fallback): {}", event);
}
}
public void logDataAccessEvent(String executionArn, String principal,
String operation, Map<String, Object> details) {
SecurityEvent event = SecurityEvent.builder()
.eventType("DATA_ACCESS")
.executionArn(executionArn)
.principal(principal)
.operation(operation)
.details(details)
.timestamp(Instant.now())
.build();
logSecurityEvent(event);
}
public List<SecurityEvent> querySecurityEvents(Instant startTime, Instant endTime,
String filterPattern) {
try {
FilterLogEventsRequest request = FilterLogEventsRequest.builder()
.logGroupName(logGroupName)
.startTime(startTime.toEpochMilli())
.endTime(endTime.toEpochMilli())
.filterPattern(filterPattern)
.build();
FilterLogEventsResponse response = cloudWatchLogsClient.filterLogEvents(request);
return response.events().stream()
.map(this::parseLogEvent)
.filter(Objects::nonNull)
.collect(Collectors.toList());
} catch (Exception e) {
throw new AuditException("Failed to query security events", e);
}
}
private String createLogMessage(SecurityEvent event) {
try {
ObjectMapper mapper = new ObjectMapper();
Map<String, Object> logData = new HashMap<>();
logData.put("eventType", event.getEventType());
logData.put("timestamp", event.getTimestamp().toString());
logData.put("principal", event.getPrincipal());
logData.put("resourceArn", event.getResourceArn());
logData.put("executionArn", event.getExecutionArn());
logData.put("operation", event.getOperation());
logData.put("severity", event.getSeverity().name());
logData.put("details", event.getDetails());
logData.put("sourceIp", event.getSourceIp());
logData.put("userAgent", event.getUserAgent());
return mapper.writeValueAsString(logData);
} catch (Exception e) {
return String.format("SecurityEvent[type=%s, principal=%s, resource=%s]",
event.getEventType(), event.getPrincipal(), event.getResourceArn());
}
}
private SecurityEvent parseLogEvent(FilteredLogEvent logEvent) {
try {
ObjectMapper mapper = new ObjectMapper();
return mapper.readValue(logEvent.message(), SecurityEvent.class);
} catch (Exception e) {
log.warn("Failed to parse log event: {}", logEvent.message());
return null;
}
}
private void initializeLogStream() {
try {
// Create log group if it doesn't exist
try {
cloudWatchLogsClient.createLogGroup(CreateLogGroupRequest.builder()
.logGroupName(logGroupName)
.build());
} catch (ResourceAlreadyExistsException e) {
// Log group already exists, this is fine
}
// Create log stream
cloudWatchLogsClient.createLogStream(CreateLogStreamRequest.builder()
.logGroupName(logGroupName)
.logStreamName(logStreamName)
.build());
} catch (Exception e) {
log.warn("Failed to initialize CloudWatch log stream", e);
}
}
}
Security Models
SecurityContext.java:
@Data
@Builder
public class SecurityContext {
private String principal;
private Set<String> roles;
private Set<String> permissions;
private String sourceIp;
private String userAgent;
private Map<String, String> attributes;
private Instant timestamp;
public boolean isAdmin() {
return roles.contains("admin") || roles.contains("Administrator");
}
public boolean hasPermission(String permission) {
return permissions.contains(permission) || isAdmin();
}
public boolean hasRole(String role) {
return roles.contains(role) || isAdmin();
}
}
@Data
@Builder
public class SecurityEvent {
private String eventType;
private Instant timestamp;
private String principal;
private String resourceArn;
private String executionArn;
private String operation;
private Severity severity;
private Map<String, Object> details;
private String sourceIp;
private String userAgent;
public enum Severity {
LOW, MEDIUM, HIGH, CRITICAL
}
}
@Data
@Builder
public class SecureExecution {
private String executionArn;
private String stateMachineArn;
private String name;
private String status;
private Instant startDate;
private Instant stopDate;
private SecurityContext securityContext;
private InputValidator.ValidationResult validationResult;
}
@Data
@Builder
public class ExecutionFilter {
private Integer maxResults;
private ExecutionStatus statusFilter;
private Instant startDate;
private Instant endDate;
}
@Data
@Builder
public class HistoryFilter {
private Integer maxResults;
private boolean reverseOrder;
private Set<String> includedEventTypes;
}
Best Practices
- Use least privilege IAM roles for Step Functions executions
- Encrypt sensitive data in state machine inputs/outputs
- Validate and sanitize all inputs before processing
- Implement comprehensive audit logging for all operations
- Use VPC endpoints for private network access
- Monitor execution patterns for anomalies
- Regularly review and update security policies
Conclusion
Step Functions Security in Java provides:
- Secure execution with input validation and encryption
- Fine-grained access control using IAM policies
- Comprehensive audit logging for compliance
- Data protection through encryption and sanitization
- Monitoring and alerting for security events
This implementation ensures that Step Functions workflows are secure, compliant, and follow AWS security best practices while maintaining the flexibility and power of serverless orchestration.