diff --git a/CHANGELOG.md b/CHANGELOG.md
index fe56bb2ef..4f41d07d9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,9 @@
 ## next
 
+## 3.9.2
 - Parse JSON with `JSON.parse` instead of `MultiJson.load`, and drop the `multi_json` dependency. On Ruby 3.0 and 3.1, `multi_json` 1.19.1 (the newest those versions can install) crashes with `json` 3.x
+- Pod log retrieval now uses the task's `kubeconfig` instead of falling back to `ENV['KUBECONFIG']`
+- Reject namespaces, contexts and kinds that `kubectl` would parse as flags rather than as positional arguments
 
 ## 3.9.1
 - Test against k8s 1.34
diff --git a/lib/krane/container_logs.rb b/lib/krane/container_logs.rb
index 356ae2999..9f4d5cf77 100644
--- a/lib/krane/container_logs.rb
+++ b/lib/krane/container_logs.rb
@@ -16,8 +16,8 @@ def initialize(parent_id:, container_name:, namespace:, context:, logger:)
       @printed_latest = false
     end
 
-    def sync
-      new_logs = fetch_latest
+    def sync(kubectl)
+      new_logs = fetch_latest(kubectl)
       return unless new_logs.present?
       @lines += sort_and_deduplicate(new_logs)
     end
@@ -47,7 +47,7 @@ def printing_started?
 
     private
 
-    def fetch_latest
+    def fetch_latest(kubectl)
       cmd = ["logs", @parent_id, "--container=#{container_name}", "--timestamps"]
       cmd << if @last_timestamp.present?
         "--since-time=#{rfc3339_timestamp(@last_timestamp)}"
@@ -58,11 +58,6 @@ def fetch_latest
       out.encode('UTF-8', invalid: :replace, replace: '').split("\n")
     end
 
-    def kubectl
-      task_config = TaskConfig.new(@context, @namespace, @logger)
-      @kubectl ||= Kubectl.new(task_config: task_config, log_failure_by_default: false)
-    end
-
     def rfc3339_timestamp(time)
       time.strftime("%FT%T.%N%:z")
     end
diff --git a/lib/krane/kubernetes_resource.rb b/lib/krane/kubernetes_resource.rb
index eb6cfe460..89ebf9689 100644
--- a/lib/krane/kubernetes_resource.rb
+++ b/lib/krane/kubernetes_resource.rb
@@ -35,6 +35,11 @@ class KubernetesResource
       If you have reason to believe it will succeed, retry the deploy to continue to monitor the rollout.
       MSG
 
+    # Kubernetes kinds begin with a letter, and may contain dashes after that (the API server
+    # lowercases the kind and validates it as a DNS-1035 label). Only the first character matters
+    # here: a leading dash would make kubectl parse the kind as a flag rather than a resource kind.
+    KIND_FORMAT = /\A[A-Za-z]/
+
     ALLOWED_DEPLOY_METHOD_OVERRIDES = %w(create replace replace-force)
     DEPLOY_METHOD_OVERRIDE_ANNOTATION = "deploy-method-override"
     TIMEOUT_OVERRIDE_ANNOTATION = "timeout-override"
@@ -91,6 +96,10 @@ def validate_definition_essentials(definition)
           raise InvalidTemplateError.new("Template is missing required field 'kind'", content: debug_content)
         end
 
+        unless definition["kind"].match?(KIND_FORMAT)
+          raise InvalidTemplateError.new("Template's 'kind' must begin with a letter", content: debug_content)
+        end
+
         if definition.dig('metadata', 'name').blank? && definition.dig('metadata', 'generateName').blank?
           raise InvalidTemplateError.new("Template must specify one of 'metadata.name' or 'metadata.generateName'",
             content: debug_content)
@@ -250,7 +259,7 @@ def deploy_method_override
 
     def sync_debug_info(kubectl)
       @debug_events = fetch_events(kubectl) unless ENV[DISABLE_FETCHING_EVENT_INFO]
-      @debug_logs = fetch_debug_logs if print_debug_logs? && !ENV[DISABLE_FETCHING_LOG_INFO]
+      @debug_logs = fetch_debug_logs(kubectl) if print_debug_logs? && !ENV[DISABLE_FETCHING_LOG_INFO]
     end
 
     def debug_message(cause = nil, info_hash = {})
diff --git a/lib/krane/kubernetes_resource/daemon_set.rb b/lib/krane/kubernetes_resource/daemon_set.rb
index 7c1c367f7..1a2075378 100644
--- a/lib/krane/kubernetes_resource/daemon_set.rb
+++ b/lib/krane/kubernetes_resource/daemon_set.rb
@@ -30,9 +30,9 @@ def deploy_failed?
       observed_generation == current_generation
     end
 
-    def fetch_debug_logs
+    def fetch_debug_logs(kubectl)
       most_useful_pod = pods.find(&:deploy_failed?) || pods.find(&:deploy_timed_out?) || pods.first
-      most_useful_pod.fetch_debug_logs
+      most_useful_pod.fetch_debug_logs(kubectl)
     end
 
     def print_debug_logs?
diff --git a/lib/krane/kubernetes_resource/deployment.rb b/lib/krane/kubernetes_resource/deployment.rb
index 7c4f81d7a..92ed8fed6 100644
--- a/lib/krane/kubernetes_resource/deployment.rb
+++ b/lib/krane/kubernetes_resource/deployment.rb
@@ -29,8 +29,8 @@ def print_debug_logs?
       @latest_rs.present?
     end
 
-    def fetch_debug_logs
-      @latest_rs.fetch_debug_logs
+    def fetch_debug_logs(kubectl)
+      @latest_rs.fetch_debug_logs(kubectl)
     end
 
     def deploy_succeeded?
diff --git a/lib/krane/kubernetes_resource/pod.rb b/lib/krane/kubernetes_resource/pod.rb
index 64a93f80a..5b42677f2 100644
--- a/lib/krane/kubernetes_resource/pod.rb
+++ b/lib/krane/kubernetes_resource/pod.rb
@@ -28,12 +28,12 @@ def initialize(namespace:, context:, definition:, logger:,
             logger: logger, statsd_tags: statsd_tags)
     end
 
-    def sync(_cache)
+    def sync(cache)
       super
       raise_predates_deploy_error if exists? && unmanaged? && !deploy_started?
 
       if exists?
-        logs.sync if unmanaged?
+        logs.sync(cache.kubectl) if unmanaged?
         update_container_statuses(@instance_data["status"])
       else # reset
         @containers.each(&:reset_status)
@@ -93,8 +93,8 @@ def failure_message
       "#{phase_failure_message} #{container_problems}".strip.presence
     end
 
-    def fetch_debug_logs
-      logs.sync
+    def fetch_debug_logs(kubectl)
+      logs.sync(kubectl)
       logs
     end
 
diff --git a/lib/krane/kubernetes_resource/pod_set_base.rb b/lib/krane/kubernetes_resource/pod_set_base.rb
index 4a4843402..58eeeb8eb 100644
--- a/lib/krane/kubernetes_resource/pod_set_base.rb
+++ b/lib/krane/kubernetes_resource/pod_set_base.rb
@@ -18,7 +18,7 @@ def fetch_events(kubectl)
       own_events.merge(most_useful_pod.fetch_events(kubectl))
     end
 
-    def fetch_debug_logs
+    def fetch_debug_logs(kubectl)
       logs = Krane::RemoteLogs.new(
         logger: @logger,
         parent_id: id,
@@ -26,7 +26,7 @@ def fetch_debug_logs
         namespace: @namespace,
         context: @context
       )
-      logs.sync
+      logs.sync(kubectl)
       logs
     end
 
diff --git a/lib/krane/remote_logs.rb b/lib/krane/remote_logs.rb
index 501c8566f..770563eec 100644
--- a/lib/krane/remote_logs.rb
+++ b/lib/krane/remote_logs.rb
@@ -23,8 +23,8 @@ def empty?
       @container_logs.all?(&:empty?)
     end
 
-    def sync
-      @container_logs.each(&:sync)
+    def sync(kubectl)
+      @container_logs.each { |cl| cl.sync(kubectl) }
     end
 
     def print_latest
diff --git a/lib/krane/resource_cache.rb b/lib/krane/resource_cache.rb
index baba9f8a8..c796218f7 100644
--- a/lib/krane/resource_cache.rb
+++ b/lib/krane/resource_cache.rb
@@ -4,6 +4,8 @@
 
 module Krane
   class ResourceCache
+    attr_reader :kubectl
+
     delegate :namespace, :context, :logger, to: :@task_config
 
     def initialize(task_config)
diff --git a/lib/krane/task_config_validator.rb b/lib/krane/task_config_validator.rb
index 8384f9076..30dcd536d 100644
--- a/lib/krane/task_config_validator.rb
+++ b/lib/krane/task_config_validator.rb
@@ -9,6 +9,10 @@ class TaskConfigValidator
       validate_namespace_exists
     ).freeze
 
+    # Kubernetes namespaces are DNS-1123 labels. Anything else, in particular a value starting with
+    # a dash, would be parsed by kubectl as a flag rather than as a resource name.
+    NAMESPACE_FORMAT = /\A[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?\z/
+
     delegate :context, :namespace, :logger, to: :@task_config
 
     def initialize(task_config, kubectl, kubeclient_builder, only: nil)
@@ -44,6 +48,12 @@ def validate_context_exists_in_kubeconfig
         return @errors << "Context can not be blank"
       end
 
+      # A context is passed to kubectl as a bare positional argument; a leading dash would make
+      # kubectl parse it as a flag. Context names are otherwise too varied to constrain further.
+      if context.start_with?("-")
+        return @errors << "Context can not start with a dash"
+      end
+
       _, err, st = @kubectl.run("config", "get-contexts", context, "-o", "name",
         use_namespace: false, use_context: false, log_failure: false, attempts: 2)
 
@@ -70,6 +80,10 @@ def validate_namespace_exists
         return @errors << "Namespace can not be blank"
       end
 
+      unless namespace.match?(NAMESPACE_FORMAT)
+        return @errors << "Namespace is not a valid Kubernetes namespace name"
+      end
+
       _, err, st = @kubectl.run("get", "namespace", "-o", "name", namespace,
         use_namespace: false, log_failure: false, attempts: 3)
 
diff --git a/lib/krane/version.rb b/lib/krane/version.rb
index e723a1fca..f4ef84bf3 100644
--- a/lib/krane/version.rb
+++ b/lib/krane/version.rb
@@ -1,4 +1,4 @@
 # frozen_string_literal: true
 module Krane
-  VERSION = "3.9.1"
+  VERSION = "3.9.2"
 end
diff --git a/test/integration/krane_deploy_test.rb b/test/integration/krane_deploy_test.rb
index 85392ad66..6e9de5d78 100644
--- a/test/integration/krane_deploy_test.rb
+++ b/test/integration/krane_deploy_test.rb
@@ -1324,7 +1324,6 @@ def test_cronjobs_can_be_deployed
   def test_jobs_can_fail
     fixtures = deploy_fixtures("hello-cloud", subset: ["job.yml"]) do |f|
       spec = f["job.yml"]["Job"].first["spec"]
-      spec["backoffLimit"] = 1
       spec["activeDeadlineSeconds"] = 1
       spec["template"]["spec"]["containers"].first["command"] = %w(/not/a/command)
     end
@@ -1334,8 +1333,8 @@ def test_jobs_can_fail
       "Deploying Job/hello-job (timeout: 600s)",
       "Result: FAILURE",
       "Job/hello-job: FAILED",
+      "DeadlineExceeded (Job was active longer than specified deadline)",
       "Final status: Failed",
-      *(%r{\[Job/hello-job\]\tDeadlineExceeded: Job was active longer than specified deadline \(\d+ events\)} if ENV['CI'] == 'true'),
     ])
   end
 
diff --git a/test/unit/container_logs_test.rb b/test/unit/container_logs_test.rb
index cdbc39ca7..fc72d00b5 100644
--- a/test/unit/container_logs_test.rb
+++ b/test/unit/container_logs_test.rb
@@ -13,14 +13,27 @@ def setup
     )
   end
 
+  def test_sync_uses_the_kubectl_it_is_given_and_builds_no_client_of_its_own
+    # Building its own Kubectl is what made log reads fall back to ENV['KUBECONFIG']
+    # instead of the credentials the task was configured with.
+    Krane::Kubectl.expects(:new).never
+    given_kubectl = mock('kubectl')
+    given_kubectl.expects(:run).with('logs', 'pod/pod-123-456', '--container=A', '--timestamps',
+      '--tail=25', log_failure: false).returns([logs_response_1, "", ""])
+
+    @logs.sync(given_kubectl)
+
+    assert_equal(generate_log_messages(1..3), @logs.lines)
+  end
+
   def test_sync_deduplicates_logs_emitted_fractional_seconds_apart
     Krane::Kubectl.any_instance.stubs(:run)
       .returns([logs_response_1, "", ""])
       .then.returns([logs_response_2, "", ""])
       .then.returns([logs_response_3, "", ""])
-    @logs.sync
-    @logs.sync
-    @logs.sync
+    @logs.sync(kubectl)
+    @logs.sync(kubectl)
+    @logs.sync(kubectl)
 
     assert_equal(generate_log_messages(1..15), @logs.lines)
   end
@@ -30,9 +43,9 @@ def test_sync_handles_cycles_where_no_new_logs_available
       .returns([logs_response_1, "", ""])
       .then.returns(["", "", ""])
       .then.returns([logs_response_2, "", ""])
-    @logs.sync
-    @logs.sync
-    @logs.sync
+    @logs.sync(kubectl)
+    @logs.sync(kubectl)
+    @logs.sync(kubectl)
 
     assert_equal(generate_log_messages(1..10), @logs.lines)
   end
@@ -40,7 +53,7 @@ def test_sync_handles_cycles_where_no_new_logs_available
   def test_empty_delegated_to_lines
     Krane::Kubectl.any_instance.stubs(:run).returns([logs_response_1, "", ""])
     assert_predicate(@logs, :empty?)
-    @logs.sync
+    @logs.sync(kubectl)
     refute_predicate(@logs, :empty?)
   end
 
@@ -49,7 +62,7 @@ def test_print_latest_and_print_all_output_the_correct_chunks
       .returns([logs_response_1, "", ""])
       .then.returns([logs_response_2, "", ""])
 
-    @logs.sync
+    @logs.sync(kubectl)
     @logs.print_latest
     assert_logs_match_all(generate_log_messages(1..3), in_order: true)
 
@@ -58,7 +71,7 @@ def test_print_latest_and_print_all_output_the_correct_chunks
     assert_logs_match_all(generate_log_messages(1..3), in_order: true)
 
     reset_logger
-    @logs.sync
+    @logs.sync(kubectl)
     @logs.print_latest
     assert_logs_match_all(generate_log_messages(4..10), in_order: true)
     refute_logs_match("Line 3")
@@ -70,7 +83,7 @@ def test_print_latest_and_print_all_output_the_correct_chunks
 
   def test_print_latest_supports_prefixing
     Krane::Kubectl.any_instance.stubs(:run).returns([logs_response_1, "", ""])
-    @logs.sync
+    @logs.sync(kubectl)
     expected = [
       "[A]  Line 1",
       "[A]  Line 2",
@@ -87,8 +100,8 @@ def test_logs_without_timestamps_are_not_deduped
       .returns([logs_response_1_with_anomaly, "", ""])
       .then.returns([logs_response_2_with_anomaly, "", ""])
 
-    @logs.sync
-    @logs.sync
+    @logs.sync(kubectl)
+    @logs.sync(kubectl)
     @logs.print_all
     assert_logs_match_all([
       "No timestamp", # moved to start of batch 1
@@ -106,9 +119,9 @@ def test_deduplication_works_when_exact_same_batch_is_returned_more_than_once
       .then.returns([logs_response_1, "", ""])
       .then.returns([logs_response_2, "", ""])
 
-    @logs.sync
-    @logs.sync
-    @logs.sync
+    @logs.sync(kubectl)
+    @logs.sync(kubectl)
+    @logs.sync(kubectl)
 
     @logs.print_all
     assert_logs_match_all(generate_log_messages(1..10), in_order: true)
@@ -140,7 +153,7 @@ def test_deduplication_works_when_last_line_is_out_of_order
       .returns([regression_data, "", ""]).times(12)
 
     12.times do
-      @logs.sync
+      @logs.sync(kubectl)
       @logs.print_latest
     end
 
@@ -154,6 +167,10 @@ def test_deduplication_works_when_last_line_is_out_of_order
 
   private
 
+  def kubectl
+    @kubectl ||= Krane::Kubectl.new(task_config: task_config, log_failure_by_default: false)
+  end
+
   def generate_log_messages(range)
     range.map { |i| "Line #{i}" }
   end
diff --git a/test/unit/krane/kubernetes_resource/pod_test.rb b/test/unit/krane/kubernetes_resource/pod_test.rb
index 537d1b36c..faaafc6da 100644
--- a/test/unit/krane/kubernetes_resource/pod_test.rb
+++ b/test/unit/krane/kubernetes_resource/pod_test.rb
@@ -324,6 +324,18 @@ def test_deploy_failed_is_false_for_disappeared_managed_pods
     assert_nil(pod.failure_message)
   end
 
+  def test_debug_logs_are_fetched_with_the_kubectl_the_caller_supplies
+    # The caller's kubectl carries the task's kubeconfig; building a fresh one here
+    # would silently fall back to ENV['KUBECONFIG'].
+    pod = Krane::Pod.new(namespace: 'test', context: 'nope', definition: build_pod_template,
+      logger: @logger, deploy_started_at: Time.now.utc)
+    Krane::Kubectl.expects(:new).never
+    given_kubectl = mock('kubectl')
+    given_kubectl.expects(:run).with { |*args, **| args.first == 'logs' }.returns(["", "", ""])
+
+    pod.fetch_debug_logs(given_kubectl)
+  end
+
   private
 
   def pod_spec
diff --git a/test/unit/krane/kubernetes_resource_test.rb b/test/unit/krane/kubernetes_resource_test.rb
index 75ba7cc53..ecf15b364 100644
--- a/test/unit/krane/kubernetes_resource_test.rb
+++ b/test/unit/krane/kubernetes_resource_test.rb
@@ -339,6 +339,23 @@ def test_lowercase_custom_resource_kind_does_not_raise
     )
   end
 
+  def test_build_rejects_kinds_that_kubectl_would_parse_as_flags
+    ["--server=http://attacker.example", "-oyaml"].each do |kind|
+      assert_raises_message(Krane::InvalidTemplateError, "Template's 'kind' must begin with a letter") do
+        Krane::KubernetesResource.build(namespace: "test", context: "test", logger: @logger,
+          statsd_tags: [], definition: { "kind" => kind, "metadata" => { "name" => "test" } })
+      end
+    end
+  end
+
+  def test_build_accepts_hyphenated_kinds
+    # The API server lowercases the kind and validates it as a DNS-1035 label, so dashes are legal
+    resource = Krane::KubernetesResource.build(namespace: "test", context: "test", logger: @logger,
+      statsd_tags: [], definition: { "kind" => "Cron-Tab", "metadata" => { "name" => "test" } })
+
+    assert_equal("Cron-Tab", resource.type)
+  end
+
   def test_build_handles_hardcoded_and_core_and_dynamic_objects
     # Dynamic with no rollout config
     no_config_crd = Krane::KubernetesResource.build(namespace: "test", context: "test",
diff --git a/test/unit/krane/task_config_validator_test.rb b/test/unit/krane/task_config_validator_test.rb
new file mode 100644
index 000000000..5ff48d51f
--- /dev/null
+++ b/test/unit/krane/task_config_validator_test.rb
@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+require 'test_helper'
+
+class TaskConfigValidatorUnitTest < Krane::TestCase
+  def test_namespace_that_would_be_parsed_as_a_kubectl_flag_is_rejected
+    kubectl = mock('kubectl')
+    kubectl.expects(:run).never
+
+    validator = build_validator(namespace: "--server=http://127.0.0.1:8080", kubectl: kubectl,
+      only: [:validate_namespace_exists])
+
+    assert_equal(["Namespace is not a valid Kubernetes namespace name"], validator.errors)
+  end
+
+  def test_namespace_that_is_not_a_dns_label_is_rejected
+    kubectl = mock('kubectl')
+    kubectl.expects(:run).never
+
+    ["UPPERCASE", "has spaces", "trailing-", "a" * 64].each do |namespace|
+      validator = build_validator(namespace: namespace, kubectl: kubectl, only: [:validate_namespace_exists])
+      assert_equal(["Namespace is not a valid Kubernetes namespace name"], validator.errors,
+        "expected #{namespace.inspect} to be rejected")
+    end
+  end
+
+  def test_valid_namespace_is_passed_to_kubectl
+    kubectl = mock('kubectl')
+    kubectl.expects(:run).with("get", "namespace", "-o", "name", "my-namespace-1",
+      use_namespace: false, log_failure: false, attempts: 3).returns(["", "", stub(success?: true)])
+
+    validator = build_validator(namespace: "my-namespace-1", kubectl: kubectl, only: [:validate_namespace_exists])
+
+    assert_empty(validator.errors)
+  end
+
+  def test_context_that_would_be_parsed_as_a_kubectl_flag_is_rejected
+    kubectl = mock('kubectl')
+    kubectl.expects(:run).never
+
+    validator = build_validator(context: "--kubeconfig=/tmp/evil", kubectl: kubectl,
+      only: [:validate_context_exists_in_kubeconfig])
+
+    assert_equal(["Context can not start with a dash"], validator.errors)
+  end
+
+  private
+
+  def build_validator(kubectl:, only:, context: "test-context", namespace: "test-namespace")
+    config = Krane::TaskConfig.new(context, namespace, logger)
+    Krane::TaskConfigValidator.new(config, kubectl, mock('kubeclient_builder'), only: only)
+  end
+end