valkey test container

[daric93]

No description provided.

[jbrinkman]

Issue #1: Command Injection Vulnerability in Password Handling

Location: modules/valkey/testcontainers/valkey/__init__.py, line 56

Current Code:

def with_password(self, password: str) -> "ValkeyContainer":
    self.password = password
    self.with_command(f"valkey-server --requirepass {password}")
    return self

Vulnerability Description:

The password is directly interpolated into a shell command without proper escaping. When Docker executes a string command, it passes it through a shell (/bin/sh -c), which interprets special characters before running the command.

Attack Scenarios:

1. Command Injection:

   container.with_password("secret; rm -rf /data; echo pwned")
   # Executes: valkey-server --requirepass secret; rm -rf /data; echo pwned
   # Result: THREE separate commands executed!

2. Command Substitution:

   container.with_password("$(cat /etc/passwd)")
   # Executes: valkey-server --requirepass $(cat /etc/passwd)
   # Result: Reads /etc/passwd and uses content as password

3. Broken Parsing:

   container.with_password("my secret password")
   # Executes: valkey-server --requirepass my secret password
   # Result: Only "my" is used as password, "secret" and "password" are extra args

Security Impact:

• OWASP Classification: A03:2021 - Injection
• Severity: HIGH
• Likelihood: MEDIUM (depends on password source)
• Potential Impact:
  • Arbitrary command execution within container
  • Data deletion or corruption
  • Container escape attempts
  • Privilege escalation
  • Unauthorized access to container resources

Recommended Fix:

Use list-based command arguments to bypass shell interpretation:

def with_password(self, password: str) -> "ValkeyContainer":
    self.password = password
    self.with_command(["valkey-server", "--requirepass", password])
    return self

Why This Works:

• List arguments are passed directly to the process without shell parsing
• Special characters are treated as literal strings
• Spaces and other characters work correctly
• Command injection becomes impossible

Comparison:

Method	Shell Parsing	Special Chars	Injection Risk
String (current)	✓ Yes	Interpreted	❌ HIGH
List (suggested)	✗ No	Literal	✅ SAFE

References:

• Docker best practices: https://docs.docker.com/engine/reference/builder/#cmd
• OWASP Injection: https://owasp.org/Top10/A03_2021-Injection/

[daric93]

fixed

[jbrinkman]

Deprecated Decorator Usage

Location: modules/valkey/testcontainers/valkey/__init__.py, line 113

Current Code:

@wait_container_is_ready(ValkeyNotReady)
def _connect(self) -> None:
    """Wait for Valkey to be ready by sending PING command."""
    # ... connection logic

Issue:

The @wait_container_is_ready decorator is deprecated in the codebase. Recent commits show active migration away from this pattern:

• PR feat: Add ExecWaitStrategy and migrate Postgres from deprecated decorator testcontainers/testcontainers-python#935: Postgres migrated to ExecWaitStrategy
• PR fix(minio): Use wait strategy instead of deprecated decorator testcontainers/testcontainers-python#899: Minio migrated to wait strategies
• PR fix(elasticsearch): Use wait strategy instead of deprecated decorator testcontainers/testcontainers-python#915: Elasticsearch migrated to wait strategies

Evidence from Codebase:

# From pytest.ini_options filterwarnings:
"ignore:The @wait_container_is_ready decorator is deprecated.*:DeprecationWarning"

Recommended Fix:

Migrate to modern wait strategy pattern:

from testcontainers.core.wait_strategies import ExecWaitStrategy

class ValkeyContainer(DockerContainer):
    def __init__(self, image: str = "valkey/valkey:latest", port: int = 6379, **kwargs) -> None:
        super().__init__(image, **kwargs)
        self.port = port
        self.password: Optional[str] = None
        self.with_exposed_ports(self.port)
        
    def start(self) -> "ValkeyContainer":
        # Build wait strategy based on password
        if self.password:
            # Use custom wait strategy for authenticated connections
            self.waiting_for(self._create_auth_wait_strategy())
        else:
            # Use exec strategy for simple PING
            self.waiting_for(
                ExecWaitStrategy(["valkey-cli", "ping"])
            )
        super().start()
        return self

Benefits:

• Aligns with project direction
• More composable and testable
• Consistent with other modules
• Better error messages

[daric93]

fixed

[jbrinkman]

Avoid Raw Socket Usage in Examples

Location: modules/valkey/example_basic.py

Current Approach:

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.connect((host, port))
    s.sendall(b"*1\r\n$4\r\nPING\r\n")  # Raw RESP protocol
    response = s.recv(1024)

Issues:

• Requires understanding of RESP protocol
• Makes examples harder to follow
• Not how users would typically interact with Valkey
• No error handling

Recommendation:

Consider using Valkey-Glide client library for cleaner examples. Kiro should be able to easily convert these examples to Valkey-Glide

[daric93]

changed