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 753100f622556..a9393abcc57ef 100644
--- a/x-pack/plugins/fleet/common/types/models/agent_policy.ts
+++ b/x-pack/plugins/fleet/common/types/models/agent_policy.ts
@@ -61,9 +61,9 @@ export interface FullAgentPolicyInput {
 }
 
 export interface FullAgentPolicyOutputPermissions {
-  [role: string]: {
-    cluster: string[];
-    indices: Array<{
+  [packagePolicyName: string]: {
+    cluster?: string[];
+    indices?: Array<{
       names: string[];
       privileges: string[];
     }>;
diff --git a/x-pack/plugins/fleet/common/types/models/epm.ts b/x-pack/plugins/fleet/common/types/models/epm.ts
index 5551453b8975c..0ef9f8b7ace36 100644
--- a/x-pack/plugins/fleet/common/types/models/epm.ts
+++ b/x-pack/plugins/fleet/common/types/models/epm.ts
@@ -276,6 +276,7 @@ export enum RegistryDataStreamKeys {
   ingest_pipeline = 'ingest_pipeline',
   elasticsearch = 'elasticsearch',
   dataset_is_prefix = 'dataset_is_prefix',
+  permissions = 'permissions',
 }
 
 export interface RegistryDataStream {
@@ -291,6 +292,7 @@ export interface RegistryDataStream {
   [RegistryDataStreamKeys.ingest_pipeline]?: string;
   [RegistryDataStreamKeys.elasticsearch]?: RegistryElasticsearch;
   [RegistryDataStreamKeys.dataset_is_prefix]?: boolean;
+  [RegistryDataStreamKeys.permissions]?: RegistryDataStreamPermissions;
 }
 
 export interface RegistryElasticsearch {
@@ -298,6 +300,11 @@ export interface RegistryElasticsearch {
   'index_template.mappings'?: object;
 }
 
+export interface RegistryDataStreamPermissions {
+  cluster?: string[];
+  indices?: string[];
+}
+
 export type RegistryVarType = 'integer' | 'bool' | 'password' | 'text' | 'yaml' | 'string';
 export enum RegistryVarsEntryKeys {
   name = 'name',
diff --git a/x-pack/plugins/fleet/server/services/agent_policy.ts b/x-pack/plugins/fleet/server/services/agent_policy.ts
index 62b4578ab87b2..2a6036d99281e 100644
--- a/x-pack/plugins/fleet/server/services/agent_policy.ts
+++ b/x-pack/plugins/fleet/server/services/agent_policy.ts
@@ -47,6 +47,10 @@ import type {
   Output,
 } from '../../common';
 import { AgentPolicyNameExistsError, HostedAgentPolicyRestrictionRelatedError } from '../errors';
+import {
+  storedPackagePoliciesToAgentPermissions,
+  DEFAULT_PERMISSIONS,
+} from '../services/package_policies_to_agent_permissions';
 
 import { getPackageInfo } from './epm/packages';
 import { getAgentsByKuery } from './agents';
@@ -745,30 +749,49 @@ class AgentPolicyService {
           }),
     };
 
+    const permissions = (await storedPackagePoliciesToAgentPermissions(
+      soClient,
+      agentPolicy.package_policies
+    )) || { _fallback: DEFAULT_PERMISSIONS };
+
+    permissions._elastic_agent_checks = {
+      cluster: DEFAULT_PERMISSIONS.cluster,
+    };
+
+    // TODO fetch this from the elastic agent package
+    const monitoringOutput = fullAgentPolicy.agent?.monitoring.use_output;
+    const monitoringNamespace = fullAgentPolicy.agent?.monitoring.namespace;
+    if (
+      fullAgentPolicy.agent?.monitoring.enabled &&
+      monitoringNamespace &&
+      monitoringOutput &&
+      fullAgentPolicy.outputs[monitoringOutput]?.type === 'elasticsearch'
+    ) {
+      const names: string[] = [];
+      if (fullAgentPolicy.agent.monitoring.logs) {
+        names.push(`logs-elastic_agent.*-${monitoringNamespace}`);
+      }
+      if (fullAgentPolicy.agent.monitoring.metrics) {
+        names.push(`metrics-elastic_agent.*-${monitoringNamespace}`);
+      }
+
+      permissions._elastic_agent_checks.indices = [
+        {
+          names,
+          privileges: ['auto_configure', 'create_doc'],
+        },
+      ];
+    }
+
     // Only add permissions if output.type is "elasticsearch"
     fullAgentPolicy.output_permissions = Object.keys(fullAgentPolicy.outputs).reduce<
       NonNullable<FullAgentPolicy['output_permissions']>
-    >((permissions, outputName) => {
+    >((outputPermissions, 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-*',
-                'synthetics-*',
-              ],
-              privileges: ['auto_configure', 'create_doc'],
-            },
-          ],
-        };
+        outputPermissions[outputName] = permissions;
       }
-      return permissions;
+      return outputPermissions;
     }, {});
 
     // 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 2db6009270a3b..dde6459addcbc 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,
         ingest_pipeline: ingestPipeline || 'default',
         path: dataStreamPath,
-        type,
         streams,
+        ...dataStreamManifestProps,
       });
     })
   );
diff --git a/x-pack/plugins/fleet/server/services/package_policies_to_agent_permissions.test.ts b/x-pack/plugins/fleet/server/services/package_policies_to_agent_permissions.test.ts
new file mode 100644
index 0000000000000..39759a6fc9e9c
--- /dev/null
+++ b/x-pack/plugins/fleet/server/services/package_policies_to_agent_permissions.test.ts
@@ -0,0 +1,341 @@
+/*
+ * 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('./epm/packages');
+import type { SavedObjectsClientContract } from 'kibana/server';
+
+import { savedObjectsClientMock } from '../../../../../src/core/server/mocks';
+import type { PackagePolicy, RegistryDataStream } from '../types';
+
+import { getPackageInfo } from './epm/packages';
+import {
+  getDataStreamPermissions,
+  storedPackagePoliciesToAgentPermissions,
+} from './package_policies_to_agent_permissions';
+
+const getPackageInfoMock = getPackageInfo as jest.MockedFunction<typeof getPackageInfo>;
+
+describe('storedPackagePoliciesToAgentPermissions()', () => {
+  let soClient: jest.Mocked<SavedObjectsClientContract>;
+  beforeEach(() => {
+    soClient = savedObjectsClientMock.create();
+  });
+
+  it('Returns `undefined` if there are no package policies', async () => {
+    const permissions = await storedPackagePoliciesToAgentPermissions(soClient, []);
+    expect(permissions).toBeUndefined();
+  });
+
+  it('Returns the default permissions for string package policies', async () => {
+    const permissions = await storedPackagePoliciesToAgentPermissions(soClient, ['foo']);
+    expect(permissions).toMatchObject({
+      _fallback: {
+        cluster: ['monitor'],
+        indices: [
+          {
+            names: [
+              'logs-*',
+              'metrics-*',
+              'traces-*',
+              'synthetics-*',
+              '.logs-endpoint.diagnostic.collection-*',
+            ],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      },
+    });
+  });
+
+  it('Returns the default permissions if a package policy does not have a package', async () => {
+    const permissions = await storedPackagePoliciesToAgentPermissions(soClient, [
+      { name: 'foo', package: undefined } as PackagePolicy,
+    ]);
+
+    expect(permissions).toMatchObject({
+      foo: {
+        cluster: ['monitor'],
+        indices: [
+          {
+            names: [
+              'logs-*',
+              'metrics-*',
+              'traces-*',
+              'synthetics-*',
+              '.logs-endpoint.diagnostic.collection-*',
+            ],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      },
+    });
+  });
+
+  it('Returns the permissions for the enabled inputs', async () => {
+    getPackageInfoMock.mockResolvedValueOnce({
+      name: 'test-package',
+      version: '0.0.0',
+      latestVersion: '0.0.0',
+      release: 'experimental',
+      format_version: '1.0.0',
+      title: 'Test Package',
+      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: [],
+        },
+      },
+      data_streams: [
+        {
+          type: 'logs',
+          dataset: 'some-logs',
+          title: '',
+          release: '',
+          package: 'test-package',
+          path: '',
+          ingest_pipeline: '',
+          streams: [{ input: 'test-logs', title: 'Test Logs', template_path: '' }],
+        },
+        {
+          type: 'metrics',
+          dataset: 'some-metrics',
+          title: '',
+          release: '',
+          package: 'test-package',
+          path: '',
+          ingest_pipeline: '',
+          streams: [{ input: 'test-metrics', title: 'Test Logs', template_path: '' }],
+        },
+      ],
+    });
+
+    const packagePolicies: PackagePolicy[] = [
+      {
+        id: '12345',
+        name: 'test-policy',
+        namespace: 'test',
+        enabled: true,
+        package: { name: 'test-package', version: '0.0.0', title: 'Test Package' },
+        inputs: [
+          {
+            type: 'test-logs',
+            enabled: true,
+            streams: [
+              {
+                id: 'test-logs',
+                enabled: true,
+                data_stream: { type: 'logs', dataset: 'some-logs' },
+              },
+            ],
+          },
+          {
+            type: 'test-metrics',
+            enabled: false,
+            streams: [
+              {
+                id: 'test-logs',
+                enabled: false,
+                data_stream: { type: 'metrics', dataset: 'some-metrics' },
+              },
+            ],
+          },
+        ],
+        created_at: '',
+        updated_at: '',
+        created_by: '',
+        updated_by: '',
+        revision: 1,
+        policy_id: '',
+        output_id: '',
+      },
+    ];
+
+    const permissions = await storedPackagePoliciesToAgentPermissions(soClient, packagePolicies);
+    expect(permissions).toMatchObject({
+      'test-policy': {
+        indices: [
+          {
+            names: ['logs-some-logs-test'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      },
+    });
+  });
+
+  it('Returns the dataset for the compiled data_streams', async () => {
+    getPackageInfoMock.mockResolvedValueOnce({
+      name: 'test-package',
+      version: '0.0.0',
+      latestVersion: '0.0.0',
+      release: 'experimental',
+      format_version: '1.0.0',
+      title: 'Test Package',
+      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: [],
+        },
+      },
+      data_streams: [
+        {
+          type: 'logs',
+          dataset: 'some-logs',
+          title: '',
+          release: '',
+          package: 'test-package',
+          path: '',
+          ingest_pipeline: '',
+          streams: [{ input: 'test-logs', title: 'Test Logs', template_path: '' }],
+        },
+      ],
+    });
+
+    const packagePolicies: PackagePolicy[] = [
+      {
+        id: '12345',
+        name: 'test-policy',
+        namespace: 'test',
+        enabled: true,
+        package: { name: 'test-package', version: '0.0.0', title: 'Test Package' },
+        inputs: [
+          {
+            type: 'test-logs',
+            enabled: true,
+            streams: [
+              {
+                id: 'test-logs',
+                enabled: true,
+                data_stream: { type: 'logs', dataset: 'some-logs' },
+                compiled_stream: { data_stream: { dataset: 'compiled' } },
+              },
+            ],
+          },
+        ],
+        created_at: '',
+        updated_at: '',
+        created_by: '',
+        updated_by: '',
+        revision: 1,
+        policy_id: '',
+        output_id: '',
+      },
+    ];
+
+    const permissions = await storedPackagePoliciesToAgentPermissions(soClient, packagePolicies);
+    expect(permissions).toMatchObject({
+      'test-policy': {
+        indices: [
+          {
+            names: ['logs-compiled-test'],
+            privileges: ['auto_configure', 'create_doc'],
+          },
+        ],
+      },
+    });
+  });
+});
+
+describe('getDataStreamPermissions()', () => {
+  it('returns defaults for a datastream with no permissions', () => {
+    const dataStream = { type: 'logs', dataset: 'test' } as RegistryDataStream;
+    const permissions = getDataStreamPermissions(dataStream);
+
+    expect(permissions).toMatchObject({
+      names: ['logs-test-*'],
+      privileges: ['auto_configure', 'create_doc'],
+    });
+  });
+
+  it('adds the namespace to the index name', () => {
+    const dataStream = { type: 'logs', dataset: 'test' } as RegistryDataStream;
+    const permissions = getDataStreamPermissions(dataStream, 'namespace');
+
+    expect(permissions).toMatchObject({
+      names: ['logs-test-namespace'],
+      privileges: ['auto_configure', 'create_doc'],
+    });
+  });
+
+  it('appends a wildcard if dataset is prefix', () => {
+    const dataStream = {
+      type: 'logs',
+      dataset: 'test',
+      dataset_is_prefix: true,
+    } as RegistryDataStream;
+    const permissions = getDataStreamPermissions(dataStream, 'namespace');
+
+    expect(permissions).toMatchObject({
+      names: ['logs-test.*-namespace'],
+      privileges: ['auto_configure', 'create_doc'],
+    });
+  });
+
+  it('prepends a dot if datastream is hidden', () => {
+    const dataStream = {
+      type: 'logs',
+      dataset: 'test',
+      hidden: true,
+    } as RegistryDataStream;
+    const permissions = getDataStreamPermissions(dataStream, 'namespace');
+
+    expect(permissions).toMatchObject({
+      names: ['.logs-test-namespace'],
+      privileges: ['auto_configure', 'create_doc'],
+    });
+  });
+
+  it('uses custom permissions if they are present in the datastream', () => {
+    const dataStream = {
+      type: 'logs',
+      dataset: 'test',
+      permissions: { indices: ['read', 'write'] },
+    } as RegistryDataStream;
+    const permissions = getDataStreamPermissions(dataStream, 'namespace');
+
+    expect(permissions).toMatchObject({
+      names: ['logs-test-namespace'],
+      privileges: ['read', 'write'],
+    });
+  });
+});
diff --git a/x-pack/plugins/fleet/server/services/package_policies_to_agent_permissions.ts b/x-pack/plugins/fleet/server/services/package_policies_to_agent_permissions.ts
new file mode 100644
index 0000000000000..bd73b88e7c893
--- /dev/null
+++ b/x-pack/plugins/fleet/server/services/package_policies_to_agent_permissions.ts
@@ -0,0 +1,152 @@
+/*
+ * 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 'kibana/server';
+
+import type { FullAgentPolicyOutputPermissions, RegistryDataStreamPermissions } from '../../common';
+import { getPackageInfo } from '../../server/services/epm/packages';
+
+import type { PackagePolicy } from '../types';
+
+export const DEFAULT_PERMISSIONS = {
+  cluster: ['monitor'],
+  indices: [
+    {
+      names: [
+        'logs-*',
+        'metrics-*',
+        'traces-*',
+        'synthetics-*',
+        '.logs-endpoint.diagnostic.collection-*',
+      ],
+      privileges: ['auto_configure', 'create_doc'],
+    },
+  ],
+};
+
+export async function storedPackagePoliciesToAgentPermissions(
+  soClient: SavedObjectsClientContract,
+  packagePolicies: string[] | PackagePolicy[]
+): Promise<FullAgentPolicyOutputPermissions | undefined> {
+  if (packagePolicies.length === 0) {
+    return;
+  }
+
+  // I'm not sure what permissions to return for this case, so let's return the defaults
+  if (typeof packagePolicies[0] === 'string') {
+    return { _fallback: DEFAULT_PERMISSIONS };
+  }
+
+  const permissionEntries = (packagePolicies as PackagePolicy[]).map<Promise<[string, any]>>(
+    async (packagePolicy) => {
+      if (!packagePolicy.package) {
+        return [packagePolicy.name, DEFAULT_PERMISSIONS];
+      }
+
+      const pkg = await getPackageInfo({
+        savedObjectsClient: soClient,
+        pkgName: packagePolicy.package.name,
+        pkgVersion: packagePolicy.package.version,
+      });
+
+      if (!pkg.data_streams || pkg.data_streams.length === 0) {
+        return [packagePolicy.name, undefined];
+      }
+
+      let dataStreamsForPermissions: DataStreamMeta[];
+
+      switch (pkg.name) {
+        case 'endpoint':
+          // - Endpoint doesn't store the `data_stream` metadata in
+          // `packagePolicy.inputs`, so we will use _all_ data_streams from the
+          // package.
+          dataStreamsForPermissions = pkg.data_streams;
+          break;
+
+        case 'apm':
+          // - APM doesn't store the `data_stream` metadata in
+          //   `packagePolicy.inputs`, so we will use _all_ data_streams from
+          //   the package.
+          dataStreamsForPermissions = pkg.data_streams;
+          break;
+
+        default:
+          // - Normal packages store some of the `data_stream` metadata in
+          //   `packagePolicy.inputs[].streams[].data_stream`
+          // - The rest of the metadata needs to be fetched from the
+          //   `data_stream` object in the package. The link is
+          //   `packagePolicy.inputs[].type == pkg.data_streams.streams[].input`
+          // - Some packages (custom logs) have a compiled dataset, stored in
+          //   `input.streams.compiled_stream.data_stream.dataset`
+          dataStreamsForPermissions = packagePolicy.inputs
+            .filter((i) => i.enabled)
+            .flatMap((input) => {
+              if (!input.streams) {
+                return [];
+              }
+
+              const dataStreams_: DataStreamMeta[] = [];
+
+              input.streams
+                .filter((s) => s.enabled)
+                .forEach((stream) => {
+                  if (!('data_stream' in stream)) {
+                    return;
+                  }
+
+                  const ds = {
+                    type: stream.data_stream.type,
+                    dataset:
+                      stream.compiled_stream?.data_stream?.dataset ?? stream.data_stream.dataset,
+                  };
+
+                  dataStreams_.push(ds);
+                });
+
+              return dataStreams_;
+            });
+      }
+
+      return [
+        packagePolicy.name,
+        {
+          indices: dataStreamsForPermissions.map((ds) =>
+            getDataStreamPermissions(ds, packagePolicy.namespace)
+          ),
+        },
+      ];
+    }
+  );
+
+  return Object.fromEntries(await Promise.all(permissionEntries));
+}
+
+interface DataStreamMeta {
+  type: string;
+  dataset: string;
+  dataset_is_prefix?: boolean;
+  hidden?: boolean;
+  permissions?: RegistryDataStreamPermissions;
+}
+
+export function getDataStreamPermissions(dataStream: DataStreamMeta, namespace: string = '*') {
+  let index = `${dataStream.type}-${dataStream.dataset}`;
+
+  if (dataStream.dataset_is_prefix) {
+    index = `${index}.*`;
+  }
+
+  if (dataStream.hidden) {
+    index = `.${index}`;
+  }
+
+  index += `-${namespace}`;
+
+  return {
+    names: [index],
+    privileges: dataStream.permissions?.indices || ['auto_configure', 'create_doc'],
+  };
+}