diff --git a/apps/common/utils/common.py b/apps/common/utils/common.py
index 6aaebaab1..6662e0f8b 100644
--- a/apps/common/utils/common.py
+++ b/apps/common/utils/common.py
@@ -10,7 +10,7 @@ from functools import wraps
 import time
 import ipaddress
 import psutil
-
+from typing import Iterable
 
 UUID_PATTERN = re.compile(r'\w{8}(-\w{4}){3}-\w{12}')
 ipip_db = None
@@ -293,4 +293,3 @@ def unique(objects, key=None):
         if v not in seen:
             seen[v] = obj
     return list(seen.values())
-
diff --git a/apps/perms/notifications.py b/apps/perms/notifications.py
new file mode 100644
index 000000000..8d93369ac
--- /dev/null
+++ b/apps/perms/notifications.py
@@ -0,0 +1,310 @@
+from datetime import datetime
+from urllib.parse import urljoin
+
+from django.utils.translation import ugettext as _
+from django.conf import settings
+
+from common.utils import reverse, get_request_ip_or_data, get_request_user_agent, lazyproperty
+from notifications.notifications import UserMessage, SystemMessage
+
+
+class AssetPermWillExpireMsg(UserMessage):
+    def __init__(self, user, assets):
+        super().__init__(user)
+        self.assets = assets
+
+    def get_html_msg(self) -> dict:
+        user = self.user
+        subject = _('Assets may expire')
+
+        assets_text = ','.join(str(asset) for asset in self.assets)
+        message = _("""
+            Hello %(name)s:
+            <br>
+            Your permissions for the following assets may expire in three days:
+            <br>
+            %(assets)s
+            <br>
+            Please contact the administrator
+        """) % {
+            'name': user.name,
+            'assets': assets_text
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+    def get_text_msg(self) -> dict:
+        user = self.user
+        subject = _('Assets may expire')
+        assets_text = ','.join(str(asset) for asset in self.assets)
+
+        message = _("""
+Hello %(name)s:
+\n
+Your permissions for the following assets may expire in three days:
+\n
+%(assets)s
+\n
+Please contact the administrator
+        """) % {
+            'name': user.name,
+            'assets': assets_text
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+
+class AssetPermWillExpireForOrgAdminMsg(UserMessage):
+    def __init__(self, user, perms, org):
+        super().__init__(user)
+        self.perms = perms
+        self.org = org
+
+    def get_html_msg(self) -> dict:
+        user = self.user
+        subject = _('Asset permission will expired')
+        perms_text = ','.join(str(perm) for perm in self.perms)
+
+        message = _("""
+            Hello %(name)s:
+            <br>
+            The following asset permissions of organization %(org) will expire in three days
+            <br>
+            %(perms)s
+        """) % {
+            'name': user.name,
+            'org': self.org,
+            'perms': perms_text
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+    def get_text_msg(self) -> dict:
+        user = self.user
+        subject = _('Asset permission will expired')
+        perms_text = ','.join(str(perm) for perm in self.perms)
+
+        message = _("""
+Hello %(name)s:
+\n
+The following asset permissions of organization %(org) will expire in three days
+\n
+%(perms)s
+        """) % {
+            'name': user.name,
+            'org': self.org,
+            'perms': perms_text
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+
+class AssetPermWillExpireForAdminMsg(UserMessage):
+    def __init__(self, user, org_perm_mapper: dict):
+        super().__init__(user)
+        self.org_perm_mapper = org_perm_mapper
+
+    def get_html_msg(self) -> dict:
+        user = self.user
+        subject = _('Asset permission will expired')
+
+        content = ''
+        for org, perms in self.org_perm_mapper.items():
+            content += f'<br> Orgnization: {org} <br> Permissions: {",".join(str(perm) for perm in perms)} <br>'
+
+        message = _("""
+            Hello %(name)s:
+            <br>
+            The following asset permissions will expire in three days
+            <br>
+            %(content)s
+        """) % {
+            'name': user.name,
+            'content': content,
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+    def get_text_msg(self) -> dict:
+        user = self.user
+        subject = _('Asset permission will expired')
+
+        content = ''
+        for org, perms in self.org_perm_mapper.items():
+            content += f'\n Orgnization: {org} \n Permissions: {perms} \n'
+
+        message = _("""
+Hello %(name)s:
+\n
+The following asset permissions of organization %(org) will expire in three days
+\n
+%(content)s
+        """) % {
+            'name': user.name,
+            'content': content,
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+
+class AppPermWillExpireMsg(UserMessage):
+    def __init__(self, user, apps):
+        super().__init__(user)
+        self.apps = apps
+
+    def get_html_msg(self) -> dict:
+        user = self.user
+        subject = _('Applications may expire')
+
+        apps_text = ','.join(str(app) for app in self.apps)
+        message = _("""
+            Hello %(name)s:
+            <br>
+            Your permissions for the following applications may expire in three days:
+            <br>
+            %(apps)s
+            <br>
+            Please contact the administrator
+        """) % {
+            'name': user.name,
+            'apps': apps_text
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+    def get_text_msg(self) -> dict:
+        user = self.user
+        subject = _('Applications may expire')
+        apps_text = ','.join(str(app) for app in self.apps)
+
+        message = _("""
+Hello %(name)s:
+\n
+Your permissions for the following applications may expire in three days:
+\n
+%(apps)s
+\n
+Please contact the administrator
+        """) % {
+            'name': user.name,
+            'apps': apps_text
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+
+class AppPermWillExpireForOrgAdminMsg(UserMessage):
+    def __init__(self, user, perms, org):
+        super().__init__(user)
+        self.perms = perms
+        self.org = org
+
+    def get_html_msg(self) -> dict:
+        user = self.user
+        subject = _('Application permission will expired')
+        perms_text = ','.join(str(perm) for perm in self.perms)
+
+        message = _("""
+            Hello %(name)s:
+            <br>
+            The following application permissions of organization %(org) will expire in three days
+            <br>
+            %(perms)s
+        """) % {
+            'name': user.name,
+            'org': self.org,
+            'perms': perms_text
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+    def get_text_msg(self) -> dict:
+        user = self.user
+        subject = _('Application permission will expired')
+        perms_text = ','.join(str(perm) for perm in self.perms)
+
+        message = _("""
+Hello %(name)s:
+\n
+The following application permissions of organization %(org) will expire in three days
+\n
+%(perms)s
+        """) % {
+            'name': user.name,
+            'org': self.org,
+            'perms': perms_text
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+
+class AppPermWillExpireForAdminMsg(UserMessage):
+    def __init__(self, user, org_perm_mapper: dict):
+        super().__init__(user)
+        self.org_perm_mapper = org_perm_mapper
+
+    def get_html_msg(self) -> dict:
+        user = self.user
+        subject = _('Application permission will expired')
+
+        content = ''
+        for org, perms in self.org_perm_mapper.items():
+            content += f'<br>Orgnization: {org} <br> Permissions: {",".join(str(perm) for perm in perms)} <br>'
+
+        message = _("""
+            Hello %(name)s:
+            <br>
+            The following application permissions will expire in three days
+            <br>
+            %(content)s
+        """) % {
+            'name': user.name,
+            'content': content,
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
+
+    def get_text_msg(self) -> dict:
+        user = self.user
+        subject = _('Application permission will expired')
+
+        content = ''
+        for org, perms in self.org_perm_mapper.items():
+            content += f'\n Orgnization: {org} \n Permissions: {perms} \n'
+
+        message = _("""
+Hello %(name)s:
+\n
+The following application permissions of organization %(org) will expire in three days
+\n
+%(content)s
+        """) % {
+            'name': user.name,
+            'content': content,
+        }
+        return {
+            'subject': subject,
+            'message': message
+        }
diff --git a/apps/perms/tasks.py b/apps/perms/tasks.py
index 6490c4ecd..a5ab4992c 100644
--- a/apps/perms/tasks.py
+++ b/apps/perms/tasks.py
@@ -1,14 +1,22 @@
 # ~*~ coding: utf-8 ~*~
 from __future__ import absolute_import, unicode_literals
 from datetime import timedelta
+from collections import defaultdict
 
 from django.db.transaction import atomic
 from django.conf import settings
 from celery import shared_task
+
+from users.models import User
+from orgs.utils import tmp_to_root_org
 from common.utils import get_logger
 from common.utils.timezone import now, dt_formater, dt_parser
 from ops.celery.decorator import register_as_period_task
-from perms.models import AssetPermission
+from perms.notifications import (
+    AssetPermWillExpireMsg, AssetPermWillExpireForOrgAdminMsg, AssetPermWillExpireForAdminMsg,
+    AppPermWillExpireMsg, AppPermWillExpireForOrgAdminMsg, AppPermWillExpireForAdminMsg,
+)
+from perms.models import AssetPermission, ApplicationPermission
 from perms.utils.asset.user_permission import UserGrantedTreeRefreshController
 
 logger = get_logger(__file__)
@@ -17,6 +25,7 @@ logger = get_logger(__file__)
 @register_as_period_task(interval=settings.PERM_EXPIRED_CHECK_PERIODIC)
 @shared_task()
 @atomic()
+@tmp_to_root_org()
 def check_asset_permission_expired():
     """
     这里的任务要足够短，不要影响周期任务
@@ -45,3 +54,82 @@ def check_asset_permission_expired():
     asset_perm_ids = list(asset_perm_ids)
     logger.info(f'>>> checking {start} to {end} have {asset_perm_ids} expired')
     UserGrantedTreeRefreshController.add_need_refresh_by_asset_perm_ids_cross_orgs(asset_perm_ids)
+
+
+@register_as_period_task(crontab='0 10 * * *')
+@shared_task()
+@atomic()
+@tmp_to_root_org()
+def check_asset_permission_will_expired():
+    start = now()
+    end = start + timedelta(days=3)
+
+    user_asset_mapper = defaultdict(set)
+    org_perm_mapper = defaultdict(set)
+
+    asset_perms = AssetPermission.objects.filter(
+        date_expired__gte=start, date_expired__lte=end
+    ).distinct()
+
+    for asset_perm in asset_perms:
+        # 资产授权按照组织分类
+        org_perm_mapper[asset_perm.org].add(asset_perm)
+
+        # 计算每个用户即将过期的资产
+        users = asset_perm.get_all_users()
+        assets = asset_perm.get_all_assets()
+
+        for u in users:
+            user_asset_mapper[u].update(assets)
+
+    for user, assets in user_asset_mapper.items():
+        AssetPermWillExpireMsg(user, assets).publish_async()
+
+    admins = User.objects.filter(role=User.ROLE.ADMIN)
+
+    if org_perm_mapper:
+        for admin in admins:
+            AssetPermWillExpireForAdminMsg(admin, org_perm_mapper).publish_async()
+
+    for org, perms in org_perm_mapper.items():
+        org_admins = org.admins.exclude(role=User.ROLE.ADMIN)
+        for org_admin in org_admins:
+            AssetPermWillExpireForOrgAdminMsg(org_admin, perms, org).publish_async()
+
+
+@register_as_period_task(crontab='0 10 * * *')
+@shared_task()
+@atomic()
+@tmp_to_root_org()
+def check_app_permission_will_expired():
+    start = now()
+    end = start + timedelta(days=3)
+
+    app_perms = ApplicationPermission.objects.filter(
+        date_expired__gte=start, date_expired__lte=end
+    ).distinct()
+
+    user_app_mapper = defaultdict(set)
+    org_perm_mapper = defaultdict(set)
+
+    for app_perm in app_perms:
+        org_perm_mapper[app_perm.org].add(app_perm)
+
+        users = app_perm.get_all_users()
+        apps = app_perm.applications.all()
+        for u in users:
+            user_app_mapper[u].update(apps)
+
+    for user, apps in user_app_mapper.items():
+        AppPermWillExpireMsg(user, apps).publish_async()
+
+    admins = User.objects.filter(role=User.ROLE.ADMIN)
+
+    if org_perm_mapper:
+        for admin in admins:
+            AppPermWillExpireForAdminMsg(admin, org_perm_mapper).publish_async()
+
+    for org, perms in org_perm_mapper.items():
+        org_admins = org.admins.exclude(role=User.ROLE.ADMIN)
+        for org_admin in org_admins:
+            AppPermWillExpireForOrgAdminMsg(org_admin, perms, org).publish_async()