信頼はずっと、挑戦はもっと。

お問い合わせ
TEL:03-3496-3888

BLOG コアテックの社員ブログ (毎週月曜~金曜更新中)

LIST OF ARTICLES

記事一覧

  • 画像:ブログサムネイル

    テクログ

    serverless frameworkで API GatewayのプライベートAPIを作成したい!

    プライベートAPI、便利ですね。内部でいろいろやるだけなのに外に出したくないですし。プライベートAPIをAPIGatewayで建てる場合、手動での作成や、既存で建ててあるAPIGatewayにリソースを追加することも可能ですが、serverless frameworkでlambdaのデプロイと同時にAPI GatewayのプライベートAPIを作成したいですよね。なのでやってみました。全体的にはhttps://www.serverless.com/framework/docs/providers/aws/events/apigateway/やhttps://github.com/serverless/serverless/pull/5080にあるのですが、いろいろ摘んで探していかないといけないので書いておきます。serverless.ymlが、このようになります!最低限くらいの内容です。service: internal-apigateway-test provider:   name: aws   endpointType: private   resourcePolicy:     - Effect: Allow       Principal: '*'       Action: execute-api:Invoke       Resource: arn:aws:execute-api:ap-northeast-1:<AWS account ID>:* #リソースにしぼってもいいです       Condition:         StringEquals:           'aws:sourceVpce': vpce-<あらかじめ作成してあるVPC Endpoint です>   runtime: python3.8   region: ap-northeast-1   role: ${opt:role_str, "arn:aws:iam::<AWS account ID>:role/aaaa"} functions:   main:     handler: handler.lambda_handler     name:  internal-apigateway-test     description: "(sample)内部APIGateway+簡単なjsonを返すだけlambdaの作成"     memorySize: 128     timeout: 5     events:       - http:           path: /           method: get ではよいslsライフを!
  • 画像:ブログサムネイル

    テクログ

    【nfcpyでLINE通知】ちょうど目の前にICカードリーダーとNFCチップが転がっていたので

    こんにちはじゅんすです。私は実家暮らしで兄弟がいるのですが、どうも兄が家に帰ってきても手を洗っていないっぽいのです。時期が時期なのでちゃんと手洗いうがいはしてほしいのですが、、。こりゃどうしたもんか、、、と思ったその時、ちょうど目の前にICカードリーダーとNFCチップが転がっていたので「手を洗ったらICカードリーダーにピッてしてLINEのトークルームに通知を送るシステム」を作っちゃいました。Pythonだと参考資料が多かったのでPythonで開発しました。触ったことないですけどね、、資料が多かったもんで、。なのでコードに至らない部分は多くあると思いますがご了承ください。以下にざっくりと処理の流れを記載しておきます。ICカードリーダーでNFCチップのIDm(固有ID)を取得。取得したIDmでDBからユーザー名を取得。「〇〇〇(ユーザー名)は手を洗ったよ」という内容でLINE通知。開発環境■各バージョンWindows 10 Home 64-bitPython 3.8.2libusb 1.0.23Zadig 2.5nfcpy 1.0.3MySQL 8.0.19■DBクライアントMySQL Workbench■USBドライバZadig■ICカードリーダー非接触型ICカードリーダー / ライター RC-S380(SONY製)■NFCチップNTAG213(NXP製)下準備ただICカードリーダー差してコードを書いただけじゃ動かないので以下の準備をしておきます。■libusbのインストール公式サイト(https://libusb.info/)からダウンロードできます(私の時はlibusb-1.0.23.7zでした)。ダウンロードしたファイルを解凍したら以下の操作をします。64ビット版Windowsの場合MS64\dll\libusb-1.0.dllをC:\Windows\System32にコピーMS32\dll\libusb-1.0.dllをC:\Windows\SysWOW64にコピー32ビット版Windowsの場合MS32\dll\libusb-1.0.dllをC:\Windows\System32にコピー■ドライバの適用公式サイト(https://zadig.akeo.ie/)からダウンロードできます(私の時はZadig 2.5でした)。ダウンロードしたらZadigアプリケーションを実行する前にPCにRC-S380を差します。ドライバが自動適用されたらZadigアプリケーションを実行します。表示されたウィンドウの上にあるリストボックスからNFC Port/PaSoRi 100 USBを選択します(ない場合はOptions ⇒ List All Devicesで表示されます)。Driverの項目はWinUSBにします。Replace Driverをクリックしたら完了です。■LINE Notify アクセストークンの発行公式サイト(https://notify-bot.line.me/ja/)にログインして、右上のアカウントからマイページに行きます。マイページの下部にトークンを発行するがあるのでトークン名の入力と通知したいトークルームを選択して発行します。※発行されたトークンは1度しか表示されないので注意。■DBの設定まずhand_washingという名前でDBを作成します。CREATE DATABASE hand_washing  CHARACTER SET utf8  COLLATE utf8_general_ci; 続いてNFCチップのデータを入れるnfc_tipsと、それを使うユーザーのデータを入れるusersテーブルを作成します。CREATE TABLE `hand_washing`.`nfc_tips` (  `id` INT NOT NULL AUTO_INCREMENT,  `idm` VARCHAR(50) NOT NULL COMMENT 'NFCチップの固有ID',  `user_id` INT NOT NULL COMMENT 'NFCチップを使用するユーザーのID',  `created_at` DATETIME NOT NULL,  `updated_at` DATETIME NOT NULL,  PRIMARY KEY (`id`)) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8 COMMENT = 'NFCチップデータ用のテーブル'; CREATE TABLE `hand_washing`.`users` (  `id` INT NOT NULL AUTO_INCREMENT,  `name` VARCHAR(10) NOT NULL COMMENT 'ユーザー名',  `created_at` DATETIME NOT NULL,  `updated_at` DATETIME NOT NULL,  PRIMARY KEY (`id`)) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8 COMMENT = 'NFCチップを使用するユーザー用テーブル'; テーブル作成も無事終わったらそれぞれのテーブルにデータを登録しておきます。nfc_tipsのidmには後述するnfc_reader.pyで取得したidmを登録します(ちょっと手間ですが、、登録機能を先に作っといた方がよかったなぁ)。動作確認■パッケージのインストールnfcpyとrequestsをpipでインストールします。pip install nfcpy pip install requests ■ファイルの作成以下3つのファイルを全て同階層に作成します。DBに接続してNFCチップのIDmからユーザー名を取得する処理を書いたdb_crud.py。import MySQLdb # データベース接続 connection = MySQLdb.connect(  host = 'ホスト名もしくはIP',  user = 'ユーザー名',  passwd = 'パスワード',  db = 'hand_washing',  charset='utf8' ) # カーソル生成 cursor = connection.cursor() # NFCのIDmからユーザー名を取得 def getUserNameByIdm(idm):  cursor.execute("SELECT users.name FROM nfc_tips JOIN users ON nfc_tips.user_id = users.id WHERE idm = '%s'" % idm.decode())  return cursor.fetchone()[0] LINEに通知を送る処理を書いたline_notify.py。import requests # LINE Notifyの情報 line_notify_token = 'アクセストークン' line_notify_api = 'https://notify-api.line.me/api/notify' # LINEでメッセージを送る def sendMessage(person):  message = str(person) + 'は手を洗ったよ'  payload = {'message': message}  headers = {'Authorization': 'Bearer ' + line_notify_token}  # 送信  requests.post(line_notify_api, data = payload, headers = headers) NFCチップのデータを読み込む処理を書いたnfc_reader.py。import nfc import binascii import time # 外部ファイル import db_crud import line_notify # タッチ待ち受けの1サイクル秒 TIME_CYCLE = 2.0 # タッチ待ち受けの反応インターバル秒 TIME_INTERVAL = 2.0 # タッチされてから次の待ち受けを開始するまで無効化する秒 TIME_WAIT = 5 # NFC接続リクエストのための準備 # NXP製のNFCチップのため106A(通信規格TypeA)で設定(Suicaで試したい人は212F) remoteTarget = nfc.clf.RemoteTarget("106A") print("カードをかざしてください") while True:  # USBに接続されたNFCリーダに接続してインスタンス化  with nfc.ContactlessFrontend("usb") as clf:   # タッチ待ち受け開始   # clf.sense( [リモートターゲット], [検索回数], [検索の間隔] )   target = clf.sense(remoteTarget, iterations = int(TIME_CYCLE//TIME_INTERVAL) + 1, interval = TIME_INTERVAL)   while target:    tag = nfc.tag.activate(clf, target)    if tag != None:     #固有のIDmを取り出す     idm = binascii.hexlify(tag.identifier).upper()     # ユーザー名取得     person = db_crud.getUserNameByIdm(idm)     print("読み込み成功\n")     print("タグ情報:" + str(tag))     print("IDm:" + idm.decode())     print("ユーザー名:" + person + "\n")     print("カードをかざしてください")     # LINEに通知する     line_notify.sendMessage(person)    else:     print("読み込みに失敗しました")    time.sleep(TIME_WAIT)    break 各ファイルが用意出来たらターミナルでnfc_reader.pyのある階層までいきます。以下のコマンドでnfc_reader.pyの処理を実行します。python nfc_reader.py すると以下のように表示されるのでNFCチップをICカードリーダーに2秒ほどかざします。python nfc_reader.py カードをかざしてください NFCチップのデータとユーザー名が表示されると同時にLINEに通知が送られます。python nfc_reader.py カードをかざしてください タグ情報:〇〇〇〇〇〇 IDm:〇〇〇〇〇〇 ユーザー名:〇〇〇〇〇〇 カードをかざしてください とまぁ、こんな感じで通知システムを作ってみたってお話です。ただ、これだけじゃ手を洗ったふりしてICカードリーダーにピッてする不正行為ができてしまうので、手の汚れをブラックライトで可視化してカメラで読み込む機能等と組み合わせるなどしたほうが良いかなとか思ったり思わなかったりって感じですね、、(Raspberry Piが欲しい)。皆さんも目の前に転がっていたら色々作ってみてください!ではでは!
  • 画像:ブログサムネイル

    テクログ

    wafのサンプルログを定期的にとって、あったら自動でchatworkに通知してほしいというきもち

    どうも、相変わらずいろいろ検証をしています。RDS Proxy はやくGAになってほしいですね。さて、標題の気持ちになったのでLambdaを書きました。ファイル分け?大丈夫大丈夫このくらいなら。冗長だとしてもわかりやすさ優先なのはいつもどおりです。弊社ではcloud9でのserverless frameworkでのデプロイも普通になってきましたね。Lambdaがはかどりますね。さてソースです。こちら、参考URLです。https://dev.classmethod.jp/articles/get-aws-waf-sample-logs/こちらを色々変更し、指定したwaf 指定したルールを 指定した間隔で行うようにしています。環境変数に入るだけ何個でもいけますね。あ、これはCF版ですが、LB版は上のクライアントを変えるだけだと思います。(作って運用しています。増えすぎるのであえてCF版とLB版を分けました)実行はcloudwatch eventsで設定です。slsの設定で。環境変数はこんな感じCHATWORK_API_TOKEN XXあんごうかしたトークンだよXXCHECK_LISTS [{'Name':'ばいたいめいだよ','WebAclID':'aaaa-aaaaとかだよ','WAFName':'わかりやすくするためのwafの名前です','RuleID':'aaaa-bbbbとかのルールID','RuleName':'これも基本わかりやすくするためのルール名'},]INTERVAL 11(とか数字)ROOM_ID 11111111(とかのルームID)# -*- coding: utf-8 -*- #cfのwafのサンプルログを確認し、あれば通知する import os import urllib from datetime import datetime, timedelta import boto3 import sys from base64 import b64decode import ast REGION = "ap-northeast-1" #環境変数 #Chatwork API トークン ENCRYPTED_CHATWORK_API_TOKEN = os.environ['CHATWORK_API_TOKEN'] CHATWORK_API_TOKEN = boto3.client('kms').decrypt(CiphertextBlob=b64decode(ENCRYPTED_CHATWORK_API_TOKEN))['Plaintext'] INTERVAL = int(os.environ['INTERVAL']) ROOM_ID = os.environ['ROOM_ID'] CHECK_LISTS = ast.literal_eval(os.environ['CHECK_LISTS']) #client用意 CF版だよ waf = boto3.client('waf') #### # replace datetime to string #### def time2str(x):     x['Timestamp'] = x['Timestamp'].isoformat()     return x #### #chatworkに通知 #### def postChatwork(message, room_id):   END_POINT_BASE = "https://api.chatwork.com/v2"   END_POINT      = "/rooms/"   ACTION         = "/messages"      headers = { 'X-ChatWorkToken': CHATWORK_API_TOKEN }      data  = { 'body': message }   data = urllib.parse.urlencode(data)   data = data.encode('utf-8')      # リクエストの生成と送信   post_message_url = END_POINT_BASE + END_POINT + room_id + ACTION   request = urllib.request.Request(post_message_url, data=data, method="POST", headers=headers)   with urllib.request.urlopen(request) as response:     response_body = response.read().decode("utf-8")     print(response_body) #### #dictを文字にするだけ #### def makeText(marge_logs):   text = "[info][title]CF版 WAF 検知ログ 直近"+str(INTERVAL)+"分[/title]"      for log in marge_logs:     for key, value in log.items():       text += key + "    " + value + "\n"     text += "[hr]"        text += "[/info]"        return text ### #main ### def lambda_handler(event, context):   #LISTごとに繰り返し   for CHECK_LIST in CHECK_LISTS:     WebAclId = CHECK_LIST['WebAclID']     acl = waf.get_web_acl(WebACLId=WebAclId)          WebACLName = acl['WebACL']['Name']        marge_logs = []     for rule in acl['WebACL']['Rules']:       RuleId = rule['RuleId']              if RuleId == CHECK_LIST['RuleID']:         # get sample requests         r = waf.get_sampled_requests(           MaxItems=3,           WebAclId=WebAclId,           RuleId=RuleId,           TimeWindow={             'EndTime': datetime.utcnow(),             'StartTime': datetime.utcnow() - timedelta(minutes=INTERVAL)           }         )              # check sample count         if 'SampledRequests' not in r or len(r['SampledRequests']) == 0:           #ログのみ           print("WebAclId:" + WebAclId + " RuleId:" +RuleId + " サンプルカウントなし")         else:           # replace datetime to string           logs = list(map(time2str, r['SampledRequests']))           for l in logs:             headers_dict = l['Request']['Headers']                      Host = ''             From = ''             UserAgent = ''             UserAgent2 = ''             Referer = ''             Referer2 = ''             for hd in headers_dict:               if hd['Name'] == 'Host':                 Host = hd['Value']               elif hd['Name'] == 'From':                 From = hd['Value']               elif hd['Name'] == 'User-Agent':                 UserAgent = hd['Value']               elif hd['Name'] == 'user-agent':                 UserAgent2 = hd['Value']               elif hd['Name'] == 'Referer':                 Referer = hd['Value']               elif hd['Name'] == 'referer':                 Referer2 = hd['Value']                          simpledict = {               '媒体名':CHECK_LIST['Name'],               '時間':l['Timestamp'],               'IP':'https://ipee.at/'+l['Request']['ClientIP'],               '国':l['Request']['Country'],               'URI':l['Request']['URI'],               'Headers.Host':Host,               'Headers.From':From,               'Headers.User-Agent':UserAgent + UserAgent2,               'Headers.Referer':Referer + Referer2,               '対象WAF/ルール':CHECK_LIST['WAFName']+' / '+CHECK_LIST['RuleName'],               'Method':l['Request']['Method'],             }             marge_logs.append(simpledict)               # chatwork通知           if marge_logs != []:             try:               print(marge_logs)               cw_text = makeText(marge_logs)               postChatwork(cw_text, ROOM_ID)             except Exception as e:               print(e)           else:             postChatwork("ログないよ", ROOM_ID)#基本ここはないですね ではよいWAF生活を!
  • 画像:ブログサムネイル

    テクログ

    slackコマンドっぽいことをchatworkでしたい!

    そろそろ夏も終わりですね。はやく涼しくなってほしいですね。そういう気持ちからslackコマンドっぽいことをchatworkでしたい!ということになりました。-----------------------◆本当の経緯もとからslackでのchatops的なものは結構やっていたんですが、弊社slackだけではなくchatworkも使うんですね(ぼやかしで、chatworkもWebhookがあって、slackよりも機能や設定できることは少ないけど、なんとか似たことができそうだな、となったので作りました。-----------------------◆構成chatwork→APIGateway→Tokenやコマンドを検証し、振り分けるLambdaFunction→(SNS)→コマンドの実態LambdaFunction+chatworkへ結果返答わかりやすくするためTokenやコマンドを検証し、振り分けるLambdaFunction を [振り分けLambda]コマンドの実態Lambda を [実態Lambda]とします-----------------------◆設定を簡単に解説細かい設定は書きませんが、いろいろ試行錯誤するのも経験のうちですよね。(1)APIGatewayを作成chatworkのWebhookが叩く入り口です。URLがあればひとまずOKです。パス設定とかはおまかせします。(2)[振り分けLambda]の殻を作成APIGatewayの先のLambdaを作成します。(3)APIGatewayを設定(1)と(2)をつなぎまーす叩くとレスポンスが来るだけ確認しておくのもいいですね。(4)chatworkのWebhook設定chatworkにボットアカウントを作り、そのアカウントでWebhookの設定をします参考:https://help.chatwork.com/hc/ja/articles/115000169541-Webhook%E3%82%92%E8%A8%AD%E5%AE%9A%E3%81%99%E3%82%8BURLはAPIGatewayのURLで、イベントはルームイベントにしました。(なぜなら、目に見えないところではなく、その部屋だけで使ってほしいからです。)メッセージ作成のみ、にしました。(5)[振り分けLambda]を実装ソース!だよ!何もライブラリが要らないってのを重視。なんか貼ったら改行があれなのでまあそこはいいかんじにしてください。何かあってもこちらはサンプルですのでそこはいつものとおりご了承ください。見直すと入ってきた文字整形とか無理矢理感。chatworkはクライアントごとにメッセージの形式が微妙に違ったりしてね…必要な環境変数はCHATWORK_API_TOKEN (KMSで暗号化してある、postに使うchatworkのTokenWEBHOOK_TOKEN (渡ってきたchatworkのメッセージが正しいものかを判定するためのTokenです。# -*- coding: utf-8 -*- # chatwork commanderの入り口 import boto3 import json import logging import os import sys from datetime import datetime from botocore.exceptions import ClientError import copy import urllib.request, urllib.error from base64 import b64decode, b64encode import hmac import hashlib valid_command_list = ['こまんど1','こまんど2','こまんど3'] to_str = "[To:1111111111]" usage_str = "【コマンド一覧】\n" + \ "※それぞれのコマンドの使い方については[コマンド usage]を打って確認してください\n\n" + \ "こまんど1\n" + \ "こまんど2\n" + \ "こまんど3\n" SNS_ARN_PREFIX = "arn:aws:sns:ap-northeast-1:11111111111:chat_command_" #### #メイン #### def lambda_handler(event, context):   #先にイベントは取っておく(判定とかに使うから   eventjson = json.loads(json.dumps(event))   bodyjson = json.loads(eventjson['body'])   room_id    = str(bodyjson['webhook_event']['room_id'])   message_id = str(bodyjson['webhook_event']['message_id'])   body      = str(bodyjson['webhook_event']['body'])   account_id = str(bodyjson['webhook_event']['account_id'])   #まず   #[To:1111111111]   #がないなら無視(全メッセージ流れてくるから   if not to_str in body:     print('not me.END')     return {'statusCode': 200,'body': json.dumps('not me.END')}   #毎回kms使うのがやだからここでやってます   ENCRYPTED_WEBHOOK_TOKEN = os.environ['WEBHOOK_TOKEN']   WEBHOOK_TOKEN = boto3.client('kms').decrypt(CiphertextBlob=b64decode(ENCRYPTED_WEBHOOK_TOKEN))['Plaintext']   ENCRYPTED_CHATWORK_API_TOKEN = os.environ['CHATWORK_API_TOKEN']   CHATWORK_API_TOKEN = boto3.client('kms').decrypt(CiphertextBlob=b64decode(ENCRYPTED_CHATWORK_API_TOKEN))['Plaintext']   #Tokenの確認   if check_request(event, WEBHOOK_TOKEN) == True:     print("Token OK")     #body整形     command_text = body.replace('\n', '').replace('[To:1111111]', '').replace('aaaaa さん', '').replace('aaaaaさん', '').strip()     command_args = command_text.split(" ")     #コマンドの確認     command = command_args[0]     nextStatus = checkCommandStrAndMakeNextStatus(command, room_id, account_id, message_id, command_text, CHATWORK_API_TOKEN)     #nextStatusがTrue(続行)の場合、次のlambdaをSNSで呼ぶ     #続行ではない場合は特に何もせずreturnして終わり     if nextStatus == True:       message = {"room_id"   : room_id,             "command_args": command_args,             "account_id" : account_id,             "message_id" : message_id       }       #SNS_ARN_PREFIX+command がARNになるようになっている       sendSNS(message, SNS_ARN_PREFIX+command)     return {       'statusCode': 200,       'body': json.dumps('OK')     }   else:     print("Token NG")     return {       'statusCode': 403,       'body': json.dumps('NG')     } ##----------------------- def sendSNS(message, sns_TOPIC):   sns_client = boto3.client(     'sns'   )   #メッセージ整形   message=json.dumps(message)   message=json.dumps({'default': message, 'lambda': message})   response = sns_client.publish(     TopicArn=sns_TOPIC,     Subject='/lambda',     MessageStructure='json',     Message=message   )   print(response) #### #コマンドの内容を確認してchatwork通知 #また、今後の続行ステータスを返す #### def checkCommandStrAndMakeNextStatus(command, room_id, account_id, message_id, command_text, CHATWORK_API_TOKEN):   if command == 'usage':     postChatwork(usage_str,room_id, CHATWORK_API_TOKEN)     return False#終了   else:     if not command in valid_command_list:       postChatwork("対象コマンドがありません。\n"+usage_str,room_id, CHATWORK_API_TOKEN)       return False#終了     else:       #chatworkに受付状況を通知       postChatwork(         makeReplyMessage(account_id, room_id, message_id, command_text),         room_id,         CHATWORK_API_TOKEN       )       return True#続行 #### #メッセージをつけるだけ #### def makeReplyMessage(account_id, room_id, message_id, command_text):   return "コマンド:" + command_text + "\nを受け付けました。(コマンドの実態に渡しました)\nお待ち下さい。" #### #chatworkに通知 #### def postChatwork(message, room_id, CHATWORK_API_TOKEN):   print("in postChatwork")   END_POINT_BASE = "https://api.chatwork.com/v2"   END_POINT   = "/rooms/"   ACTION     = "/messages"   headers = { 'X-ChatWorkToken': CHATWORK_API_TOKEN }   data = { 'body': message }   data = urllib.parse.urlencode(data)   data = data.encode('utf-8')   # リクエストの生成と送信   post_message_url = END_POINT_BASE + END_POINT + room_id + ACTION   request = urllib.request.Request(post_message_url, data=data, method="POST", headers=headers)   with urllib.request.urlopen(request) as response:     response_body = response.read().decode("utf-8")     print(response_body) #参考 #https://qiita.com/ikedaosushi/items/b01183f207ea1db6a8d3 def check_request(event, WEBHOOK_TOKEN):   webhook_sig = event['headers']['X-ChatWorkWebhookSignature'].encode('utf-8')   sig = build_sig(event['body'], WEBHOOK_TOKEN)   is_valid = webhook_sig == sig   return is_valid def build_sig(body, WEBHOOK_TOKEN):   token_decode = b64decode(WEBHOOK_TOKEN)   sig = hmac.new(token_decode, body.encode('utf-8'), hashlib.sha256).digest()   sig_encoded = b64encode(sig)   return sig_encoded だいたいでいえば・メッセージはルームのものがすべて来るので、処理対象メッセージなのかを判定・tokenの確認・コマンドの確認・受け付けたことを返答・SNSで次のLambdaを呼ぶ(このSNSを一定の形式に統一することによって、後ろのコマンドを増やしやすくなってます。(6)実態Lambdaを呼ぶためのSNSを作るarn:aws:sns:ap-northeast-1:11111111111:chat_command_ ってなっているようにSNSをまず作りますたとえばmake_EC2みたいなコマンドだった場合、SNS名はchat_command_make_EC2になりますね。(7)[実態Lambda]を作るここは自分でつくってね!SNSを受け取って、分解して、処理をすればいいです。終わったらchatworkに通知したり、いろいろ内部でやればいいですね。夢が広がる。(8)SNSの先に[実態Lambda]を追加まあ追加するだけですあとはchatworkでボットにはなしかけたりしてテストをしてみよう!-----------------------コマンドを増やすときも、許可するコマンドを追加して、SNS作成コマンド実態Lambda作成だけでぽんぽんふえるよ!ルームを分けることによって、重要度ごとに使える人を分けられたりもします。以上夏休みの自由研究にどうぞ。もうおわってますか。-----------------------<以下はただの一覧用の画像でーす>
  • 画像:ブログサムネイル

    テクログ

    コスト配分タグを自動でチェックしたい!

    恒例のしたいシリーズです。いまつくりました。●なにがしたいの?AWSアカウント内に複数の媒体があって、どの媒体にどれだけお金がかかっているかをチェックするために、リソースにProjectタグをつけると簡単に集計できます。でも、リソースは随時増えるし、ちゃんとProjectタグついているのかわからない。なんとかしたいという気持ち。●どうやるの?LambdaでリソースにProjectタグがついているかをチェック。+Projectタグの値もチェック(ちがうのがついてたらこまるからね)チェックしない、というのも設定したい。あと、結果はSlackに通知したりしたい。●具体的には?ソースをはりますんで見てね!EC2のタグチェックの例です。サンプルですよ。・追加ライブラリslackwebいれてください。・使っている環境変数ACCOUNT_NAME AWSアカウントの名前とかをいれます。通知用です。CHECK_TAG_NAME チェックしたいタグをいれます。今回はProjectです。CHECK_TAG_VALUES カンマ区切りで、入っていていほしいタグの値をいれます。aaa.jp,bbb.jp  みたいなのですSLACK_URL SlackのWebhookURLです。・ソース(Pythonらしさはかけらもありませんがゆるしてください# -*- coding: utf-8 -*- # EC2のタグ設定チェック import boto3 import os import slackweb import sys # アカウント名 ACCOUNT_NAME   = os.environ['ACCOUNT_NAME'] # チェックするタグ名 CHECK_TAG_NAME  = os.environ['CHECK_TAG_NAME'] # ついていてほしいタグの内容(どれかならOK) CHECK_TAG_VALUES = os.environ['CHECK_TAG_VALUES'].split(",") #slackurl SLACK_URL    = os.environ['SLACK_URL'] REGION_NAME = 'ap-northeast-1' ##################################################### #本体 ##################################################### def lambda_handler(event, context):   #タグチェック   results = checkTags()   print('checkTags results:')   print(results)   checkResult_AND_Notify(results) ##################################################### # インスタンスID一覧を取得 ##################################################### def checkTags():   instance_list = []   #ページャーで全て取得   client  = boto3.client('ec2', region_name=REGION_NAME)   paginator = client.get_paginator('describe_instances')   # Filterで止まっているものは除外   page_iterator = paginator.paginate(     Filters=[     {       'Name': 'instance-state-name',       'Values': [         'running',       ]     }   ],)   #チェックしていく   for page in page_iterator:     for reservation in page["Reservations"]:       for instance in reservation["Instances"]:         instance_id = instance["InstanceId"]         #Tagsを見て、CheckProjectSettingがNOならば除外(存在しない、またはYESなら追加         check_flag       = True         project_tag_exist_flag = False         project_tag_value   = ""         name_tag_value     = ""         Tags = instance["Tags"]         for tag in Tags:           if tag['Key'] == 'CheckProjectSetting':             if tag['Value'] == 'NO':               print('CheckProjectSetting,NO:' + instance_id)               check_flag = False           if tag['Key'] == 'Project':             project_tag_exist_flag = True             project_tag_value = tag['Value']             print('Project tag_value:' + project_tag_value)           #Nameはわかりやすさのため取ります           if tag['Key'] == 'Name':             name_tag_value = tag['Value']             print('Project tag_value:' + name_tag_value)         if check_flag == True:           print('CheckProjectSetting,YES:' + instance_id)           if project_tag_exist_flag == False:             #タグ自体がない場合は即追加             instance_id_dict = {               "instance_id"    : instance_id,               "name_tag_value"  : name_tag_value,               "project_tag_value" : project_tag_value,               "reason"      : "no_project_tag"             }             instance_list.append(instance_id_dict)           else:             #タグはあった場合、中身を確認し、なかったら一覧追加             if project_tag_value not in CHECK_TAG_VALUES:               instance_id_dict = {                 "instance_id"    : instance_id,                 "name_tag_value"  : name_tag_value,                 "project_tag_value" : project_tag_value,                 "reason"      : "project_tag_not_match"               }               instance_list.append(instance_id_dict)   return instance_list ##################################################### # 漏れがあるものを通知 ##################################################### def checkResult_AND_Notify(results):   print('checkResult_AND_Notify')   exist_no_Tag_Flag = False   res_str = ''   for result in results:     exist_no_Tag_Flag = True     res_str += '&gt;  Name:' + result['name_tag_value'] + ' ID:' + result['instance_id'] + ' 理由:' + result['reason'] +'\n'   if exist_no_Tag_Flag:     title_str = "ProjectタグをつけてないEC2インスタンスがあったよ!付けてね!\n"     slack = slackweb.Slack(url=SLACK_URL)     slack.notify(text=ACCOUNT_NAME + "のProjectタグチェックをしたよ!\n\n" + title_str + res_str)   else:     print("ProjectタグをつけてないEC2インスタンスはなかったよ!") ※Tagsを見て、CheckProjectSettingがNOならば除外やったね!もうタグがあるかどうかをチェックしなくていいよ!肝心のタグ設定は場合によりますので、手動で付けてくださいね。以上!(見栄え用の画像↓)