diff --git a/x-pack/plugins/fleet/common/types/index.ts b/x-pack/plugins/fleet/common/types/index.ts
index 1984de79a6357e..b48f99c3481848 100644
--- a/x-pack/plugins/fleet/common/types/index.ts
+++ b/x-pack/plugins/fleet/common/types/index.ts
@@ -31,6 +31,7 @@ export interface FleetConfigType {
     };
     agentPolicyRolloutRateLimitIntervalMs: number;
     agentPolicyRolloutRateLimitRequestPerInterval: number;
+    agentPolicyTightPermissions: boolean;
   };
 }
 
diff --git a/x-pack/plugins/fleet/common/types/models/agent_policy.ts b/x-pack/plugins/fleet/common/types/models/agent_policy.ts
index 439d00695a7376..d6516014ef3665 100644
--- a/x-pack/plugins/fleet/common/types/models/agent_policy.ts
+++ b/x-pack/plugins/fleet/common/types/models/agent_policy.ts
@@ -10,6 +10,7 @@ import type { DataType, ValueOf } from '../../types';
 
 import type { PackagePolicy, PackagePolicyPackage } from './package_policy';
 import type { Output } from './output';
+import type { PackagePermissions } from './epm';
 
 export type AgentPolicyStatus = typeof agentPolicyStatuses;
 
@@ -61,13 +62,7 @@ export interface FullAgentPolicyInput {
 }
 
 export interface FullAgentPolicyOutputPermissions {
-  [role: string]: {
-    cluster: string[];
-    indices: Array<{
-      names: string[];
-      privileges: string[];
-    }>;
-  };
+  [role: string]: PackagePermissions;
 }
 
 export interface FullAgentPolicy {
diff --git a/x-pack/plugins/fleet/common/types/models/epm.ts b/x-pack/plugins/fleet/common/types/models/epm.ts
index 3bc0d97d646465..65fbb300b14798 100644
--- a/x-pack/plugins/fleet/common/types/models/epm.ts
+++ b/x-pack/plugins/fleet/common/types/models/epm.ts
@@ -256,6 +256,7 @@ export enum RegistryDataStreamKeys {
   ingest_pipeline = 'ingest_pipeline',
   elasticsearch = 'elasticsearch',
   dataset_is_prefix = 'dataset_is_prefix',
+  permissions = 'permissions',
 }
 
 export interface RegistryDataStream {
@@ -271,6 +272,7 @@ export interface RegistryDataStream {
   [RegistryDataStreamKeys.ingest_pipeline]: string;
   [RegistryDataStreamKeys.elasticsearch]?: RegistryElasticsearch;
   [RegistryDataStreamKeys.dataset_is_prefix]?: boolean;
+  [RegistryDataStreamKeys.permissions]?: RegistryDataStreamPermissions;
 }
 
 export interface RegistryElasticsearch {
@@ -278,6 +280,16 @@ export interface RegistryElasticsearch {
   'index_template.mappings'?: object;
 }
 
+export interface RegistryDataStreamPermissions {
+  cluster?: string[];
+  indices?: string[];
+}
+
+export interface PackagePermissions {
+  cluster?: string[];
+  indices?: Array<{ names: string[]; privileges: string[] }>;
+}
+
 export type RegistryVarType = 'integer' | 'bool' | 'password' | 'text' | 'yaml' | 'string';
 export enum RegistryVarsEntryKeys {
   name = 'name',
diff --git a/x-pack/plugins/fleet/public/applications/fleet/mock/plugin_configuration.ts b/x-pack/plugins/fleet/public/applications/fleet/mock/plugin_configuration.ts
index 81ef6a6703c343..0026d254bd94b2 100644
--- a/x-pack/plugins/fleet/public/applications/fleet/mock/plugin_configuration.ts
+++ b/x-pack/plugins/fleet/public/applications/fleet/mock/plugin_configuration.ts
@@ -28,6 +28,7 @@ export const createConfigurationMock = (): FleetConfigType => {
       },
       agentPolicyRolloutRateLimitIntervalMs: 100,
       agentPolicyRolloutRateLimitRequestPerInterval: 1000,
+      agentPolicyTightPermissions: false,
     },
   };
 };
diff --git a/x-pack/plugins/fleet/server/index.ts b/x-pack/plugins/fleet/server/index.ts
index 0178b801f4d2fc..2e67d6ae5de08b 100644
--- a/x-pack/plugins/fleet/server/index.ts
+++ b/x-pack/plugins/fleet/server/index.ts
@@ -76,6 +76,7 @@ export const config: PluginConfigDescriptor = {
       agentPolicyRolloutRateLimitRequestPerInterval: schema.number({
         defaultValue: AGENT_POLICY_ROLLOUT_RATE_LIMIT_REQUEST_PER_INTERVAL,
       }),
+      agentPolicyTightPermissions: schema.boolean({ defaultValue: false }),
     }),
   }),
 };
diff --git a/x-pack/plugins/fleet/server/services/agent_policy.ts b/x-pack/plugins/fleet/server/services/agent_policy.ts
index 7f793a41ab9855..40c37c850e3c95 100644
--- a/x-pack/plugins/fleet/server/services/agent_policy.ts
+++ b/x-pack/plugins/fleet/server/services/agent_policy.ts
@@ -38,6 +38,7 @@ import {
   AGENT_POLICY_INDEX,
   DEFAULT_FLEET_SERVER_AGENT_POLICY,
 } from '../../common';
+import type { FullAgentPolicyOutputPermissions, PackagePermissions } from '../../common';
 import type {
   DeleteAgentPolicyResponse,
   Settings,
@@ -52,7 +53,7 @@ import {
 } from '../errors';
 import { getFullAgentPolicyKibanaConfig } from '../../common/services/full_agent_policy_kibana_config';
 
-import { getPackageInfo } from './epm/packages';
+import { getPackageInfo, getPackagePermissions } from './epm/packages';
 import { createAgentPolicyAction, getAgentsByKuery } from './agents';
 import { packagePolicyService } from './package_policy';
 import { outputService } from './output';
@@ -64,6 +65,16 @@ import { appContextService } from './app_context';
 
 const SAVED_OBJECT_TYPE = AGENT_POLICY_SAVED_OBJECT_TYPE;
 
+const DEFAULT_PERMISSIONS: PackagePermissions = {
+  cluster: ['monitor'],
+  indices: [
+    {
+      names: ['logs-*', 'metrics-*', 'traces-*', '.logs-endpoint.diagnostic.collection-*'],
+      privileges: ['auto_configure', 'create_doc'],
+    },
+  ],
+};
+
 class AgentPolicyService {
   private triggerAgentPolicyUpdatedEvent = async (
     soClient: SavedObjectsClientContract,
@@ -739,24 +750,53 @@ class AgentPolicyService {
           }),
     };
 
+    const hasTightPermissions = appContextService.getConfig()?.agents.agentPolicyTightPermissions;
+    let permissions: FullAgentPolicyOutputPermissions;
+
+    if (hasTightPermissions) {
+      permissions = Object.fromEntries(
+        await Promise.all(
+          // Original type is `string[] | PackagePolicy[]`, but TS doesn't allow to `map()` over that.
+          (agentPolicy.package_policies as Array<string | PackagePolicy>).map(
+            async (packagePolicy): Promise<[string, PackagePermissions]> => {
+              if (typeof packagePolicy === 'string' || !packagePolicy.package) {
+                return ['_fallback', DEFAULT_PERMISSIONS];
+              }
+
+              const { name, version } = packagePolicy.package;
+
+              const packagePermissions = await getPackagePermissions(
+                soClient,
+                name,
+                version,
+                packagePolicy.namespace
+              );
+
+              return packagePermissions
+                ? [packagePolicy.name, packagePermissions]
+                : ['_fallback', DEFAULT_PERMISSIONS];
+            }
+          )
+        )
+      );
+      permissions._elastic_agent_checks = {
+        cluster: DEFAULT_PERMISSIONS.cluster,
+      };
+    } else {
+      permissions = {
+        _fallback: DEFAULT_PERMISSIONS,
+      };
+    }
+
     // Only add permissions if output.type is "elasticsearch"
     fullAgentPolicy.output_permissions = Object.keys(fullAgentPolicy.outputs).reduce<
       NonNullable<FullAgentPolicy['output_permissions']>
-    >((permissions, outputName) => {
+    >((p, outputName) => {
       const output = fullAgentPolicy.outputs[outputName];
       if (output && output.type === 'elasticsearch') {
-        permissions[outputName] = {};
-        permissions[outputName]._fallback = {
-          cluster: ['monitor'],
-          indices: [
-            {
-              names: ['logs-*', 'metrics-*', 'traces-*', '.logs-endpoint.diagnostic.collection-*'],
-              privileges: ['auto_configure', 'create_doc'],
-            },
-          ],
-        };
+        p[outputName] = permissions;
       }
-      return permissions;
+      return p;
     }, {});
 
     // only add settings if not in standalone
diff --git a/x-pack/plugins/fleet/server/services/epm/archive/storage.ts b/x-pack/plugins/fleet/server/services/epm/archive/storage.ts
index 2db6009270a3b6..32d499b07ba116 100644
--- a/x-pack/plugins/fleet/server/services/epm/archive/storage.ts
+++ b/x-pack/plugins/fleet/server/services/epm/archive/storage.ts
@@ -224,24 +224,20 @@ export const getEsPackage = async (
       );
       const dataStreamManifest = safeLoad(soResDataStreamManifest.attributes.data_utf8);
       const {
-        title: dataStreamTitle,
-        release,
         ingest_pipeline: ingestPipeline,
-        type,
         dataset,
         streams: manifestStreams,
+        ...dataStreamManifestProps
       } = dataStreamManifest;
       const streams = parseAndVerifyStreams(manifestStreams, dataStreamPath);
 
       dataStreams.push({
-        dataset: dataset || `${pkgName}.${dataStreamPath}`,
-        title: dataStreamTitle,
-        release,
         package: pkgName,
+        dataset: dataset || `${pkgName}.${dataStreamPath}`,
         ingest_pipeline: ingestPipeline || 'default',
         path: dataStreamPath,
-        type,
         streams,
+        ...dataStreamManifestProps,
       });
     })
   );
diff --git a/x-pack/plugins/fleet/server/services/epm/packages/index.ts b/x-pack/plugins/fleet/server/services/epm/packages/index.ts
index c85376ef177b3f..7b54c46a74fd3a 100644
--- a/x-pack/plugins/fleet/server/services/epm/packages/index.ts
+++ b/x-pack/plugins/fleet/server/services/epm/packages/index.ts
@@ -24,6 +24,8 @@ export {
   SearchParams,
 } from './get';
 
+export { getPackagePermissions } from './permissions';
+
 export {
   BulkInstallResponse,
   IBulkInstallPackageError,
diff --git a/x-pack/plugins/fleet/server/services/epm/packages/permissions.test.ts b/x-pack/plugins/fleet/server/services/epm/packages/permissions.test.ts
new file mode 100644
index 00000000000000..21f7473afd8fff
--- /dev/null
+++ b/x-pack/plugins/fleet/server/services/epm/packages/permissions.test.ts
@@ -0,0 +1,255 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+jest.mock('./get', () => ({ getPackageInfo: jest.fn(async () => ({})) }));
+
+import type { SavedObjectsClientContract } from 'kibana/server';
+
+import { savedObjectsClientMock } from '../../../../../../../src/core/server/mocks';
+import type { PackageInfo, RegistryDataStream } from '../../../types';
+
+import { getPackagePermissions } from './permissions';
+import { getPackageInfo } from './get';
+
+const getPackageInfoMock = getPackageInfo as jest.MockedFunction<typeof getPackageInfo>;
+
+const PACKAGE = 'test_package';
+const VERSION = '1.0.0';
+
+function createFakePackage(props: Partial<PackageInfo> = {}): PackageInfo {
+  const name = PACKAGE;
+  const version = VERSION;
+
+  return {
+    name,
+    version,
+    latestVersion: version,
+    release: 'experimental',
+    format_version: '1.0.0',
+    title: name,
+    description: '',
+    icons: [],
+    owner: { github: '' },
+    status: 'not_installed',
+    assets: {
+      kibana: {
+        dashboard: [],
+        visualization: [],
+        search: [],
+        index_pattern: [],
+        map: [],
+        lens: [],
+        security_rule: [],
+        ml_module: [],
+      },
+      elasticsearch: {
+        component_template: [],
+        ingest_pipeline: [],
+        ilm_policy: [],
+        transform: [],
+        index_template: [],
+        data_stream_ilm_policy: [],
+      },
+    },
+    ...props,
+  } as PackageInfo;
+}
+
+function createFakeDataset(
+  type: 'logs' | 'metrics' | 'traces' | 'synthetics',
+  dataset: string,
+  props: Partial<RegistryDataStream> = {}
+): RegistryDataStream {
+  return {
+    type,
+    dataset,
+    title: dataset,
+    package: PACKAGE,
+    release: VERSION,
+    path: `/${dataset}/`,
+    ingest_pipeline: '',
+    ...props,
+  };
+}
+
+describe('epm/permissions', () => {
+  let soClient: jest.Mocked<SavedObjectsClientContract>;
+
+  beforeEach(() => {
+    soClient = savedObjectsClientMock.create();
+    getPackageInfoMock.mockReset();
+  });
+
+  describe('getPackagePermissions()', () => {
+    it('Returns `undefined` if package does not have datasets', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(createFakePackage());
+
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION);
+      expect(permissions).toBeUndefined();
+    });
+
+    it('Returns empty permissions if datasets are empty', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(createFakePackage({ data_streams: [] }));
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION);
+      expect(permissions).toMatchObject({ cluster: [], indices: [] });
+    });
+
+    it('Returns default permissions', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(
+        createFakePackage({
+          data_streams: [createFakeDataset('logs', 'dataset')],
+        })
+      );
+
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION);
+      expect(permissions).toMatchObject({
+        cluster: ['monitor'],
+        indices: [
+          {
+            names: ['logs-dataset-*'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      });
+    });
+
+    it('Returns default permissions for multiple datasets', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(
+        createFakePackage({
+          data_streams: [
+            createFakeDataset('logs', 'dataset1'),
+            createFakeDataset('metrics', 'dataset2'),
+          ],
+        })
+      );
+
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION);
+      expect(permissions).toMatchObject({
+        cluster: ['monitor'],
+        indices: [
+          {
+            names: ['logs-dataset1-*'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+          {
+            names: ['metrics-dataset2-*'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      });
+    });
+
+    it('Passes the namespace to the datasets', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(
+        createFakePackage({
+          data_streams: [createFakeDataset('logs', 'dataset')],
+        })
+      );
+
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION, 'test');
+      expect(permissions).toMatchObject({
+        cluster: ['monitor'],
+        indices: [
+          {
+            names: ['logs-dataset-test'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      });
+    });
+
+    it('Handles hidden datasets', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(
+        createFakePackage({
+          data_streams: [createFakeDataset('logs', 'dataset', { hidden: true })],
+        })
+      );
+
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION);
+      expect(permissions).toMatchObject({
+        cluster: ['monitor'],
+        indices: [
+          {
+            names: ['.logs-dataset-*'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      });
+    });
+
+    it('Handles prefix datasets', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(
+        createFakePackage({
+          data_streams: [createFakeDataset('logs', 'dataset', { dataset_is_prefix: true })],
+        })
+      );
+
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION, 'test');
+      expect(permissions).toMatchObject({
+        cluster: ['monitor'],
+        indices: [
+          {
+            names: ['logs-dataset-test-*'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      });
+    });
+
+    it('Aggregates cluster permissions from the different datasets', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(
+        createFakePackage({
+          data_streams: [
+            createFakeDataset('logs', 'dataset', { permissions: { cluster: ['foo', 'bar'] } }),
+            createFakeDataset('metrics', 'dataset', { permissions: { cluster: ['foo', 'baz'] } }),
+          ],
+        })
+      );
+
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION);
+      expect(permissions).toMatchObject({
+        cluster: ['foo', 'bar', 'baz'],
+        indices: [
+          {
+            names: ['logs-dataset-*'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+          {
+            names: ['metrics-dataset-*'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      });
+    });
+
+    it('Configures indices permissions for each datasetdatasets', async () => {
+      getPackageInfoMock.mockResolvedValueOnce(
+        createFakePackage({
+          data_streams: [
+            createFakeDataset('logs', 'dataset', { permissions: { indices: ['foo', 'bar'] } }),
+            createFakeDataset('metrics', 'dataset', { permissions: { indices: ['foo', 'baz'] } }),
+          ],
+        })
+      );
+
+      const permissions = await getPackagePermissions(soClient, PACKAGE, VERSION);
+      expect(permissions).toMatchObject({
+        cluster: ['monitor'],
+        indices: [
+          {
+            names: ['logs-dataset-*'],
+            privileges: ['foo', 'bar'],
+          },
+          {
+            names: ['metrics-dataset-*'],
+            privileges: ['foo', 'baz'],
+          },
+        ],
+      });
+    });
+  });
+});
diff --git a/x-pack/plugins/fleet/server/services/epm/packages/permissions.ts b/x-pack/plugins/fleet/server/services/epm/packages/permissions.ts
new file mode 100644
index 00000000000000..4b934639a0f50e
--- /dev/null
+++ b/x-pack/plugins/fleet/server/services/epm/packages/permissions.ts
@@ -0,0 +1,49 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { SavedObjectsClientContract } from 'src/core/server';
+
+import type { PackagePermissions } from '../../../../common/types';
+
+import { getPackageInfo } from './get';
+
+export async function getPackagePermissions(
+  soClient: SavedObjectsClientContract,
+  pkgName: string,
+  pkgVersion: string,
+  namespace = '*'
+): Promise<PackagePermissions | undefined> {
+  const pkg = await getPackageInfo({ savedObjectsClient: soClient, pkgName, pkgVersion });
+  if (!pkg.data_streams) {
+    return undefined;
+  }
+
+  const clusterPermissions = new Set<string>();
+  const indices: PackagePermissions['indices'] = pkg.data_streams!.map((ds) => {
+    if (ds.permissions?.cluster) {
+      ds.permissions.cluster.forEach((p) => clusterPermissions.add(p));
+    }
+
+    let index = `${ds.type}-${ds.dataset}-${namespace}`;
+    if (ds.dataset_is_prefix) {
+      index = `${index}-*`;
+    }
+    if (ds.hidden) {
+      index = `.${index}`;
+    }
+
+    return {
+      names: [index],
+      privileges: ds.permissions?.indices ?? ['auto_configure', 'create_doc'],
+    };
+  });
+
+  return {
+    cluster: clusterPermissions.size > 0 ? Array.from(clusterPermissions) : undefined,
+    indices,
+  };
+}